p2p

nexutils/p2p

^{the p2p module, part of GSF-nexutils, member of the tiny-frameworks family}

nexutils/p2p is a lightweight, high-performance JSON-RPC 2.0 over WebSocket package for Go. It is built on top of github.com/coder/websocket with automatic session management, heartbeats, and resilient client control. It is specifically designed for Peer-to-Peer (P2P) scenarios and distributed systems.

Key Features

• Delegate-Driven Architecture (NexDelegate): Decoupled, object-oriented lifecycle management and RPC dispatching replacing map-based route handlers.
• True Peer Symmetry: Every node (Node) can act simultaneously as a server (incoming connections) and as a client (outgoing connections).
• Managed Client Architecture: Decoupled, stateful auto-reconnect daemon featuring configurable Exponential Backoff & Jitter.
• Type-Safe JSON-RPC 2.0: Full support for synchronous method calls (Call) and asynchronous one-way events (Notify).
• Context-Driven: Full cancellation support and strict request timeouts to prevent goroutine leaks.
• Heartbeat & Time Sync: Integrated liveness check driven by the client role. The receiving peer automatically responds with pong and a precise UTC timestamp.
• No TLS Overhead in Code: Designed for secure internal environments or operation behind reverse proxies (e.g., Caddy), which handle TLS termination more efficiently.
• Structured Errors: Full integration with native JSON-RPC error codes and custom business error types.

Architecture & Concepts

1. Components

The package is split into the following core responsibilities:

2. Peer-to-Peer (P2P) Principle & Delegate Dispatching

Once the initial WebSocket handshake via HTTP GET is completed, the connection upgrades to the bidirectional WebSocket protocol. Inbound requests and lifecycle events bypass message queues and route maps, executing directly via the injected NexDelegate.

│   [ Node A ]                                  │   [ Node B ]

(Port :8080 / Server)                           (Port :8081 / Server)
▲                                               │
│────── ConnectToPeer("ws://nodeA/ws") ─────────│  (RoleOutbound)
│  or ConnectWithAutoReconnect(...)             │
│                                               ▼
├─────────────── Request: "auth" ───────────────┤  ──> NexDelegate.OnRequest()
├─────────────── Request: "heartbeat" ──────────┤  ──> NexDelegate.OnRequest()
├─────────────── Notify: "event" ───────────────┤  ──> NexDelegate.OnNotification()

Installation & Import

go get codeberg.org/tiny-frameworks/nexutils/p2p

import "codeberg.org/tiny-frameworks/nexutils/p2p/rpc"

Quickstart & Examples

1. Custom Delegate Implementation

Embed rpc.DefaultNexDelegate and override only the lifecycle or request hooks you need:

package main

import (
    "context"
    "fmt"
    "log"

    "codeberg.org/tiny-frameworks/nexutils/p2p/rpc"
)

type MyDelegate struct {
    rpc.DefaultNexDelegate
}

// Peer connection validation (Auth / Filtering)
func (d *MyDelegate) ValidatePeer(peer *rpc.Peer) bool {
    log.Printf("Validating peer connection from %s", peer.RemoteAddr())
    return true
}

func (d *MyDelegate) OnPeerConnected(peer *rpc.Peer) {
    log.Printf("Peer connected: %s", peer.ID)
}

func (d *MyDelegate) OnPeerDisconnected(peer *rpc.Peer, err error) {
    log.Printf("Peer disconnected: %s (Reason: %v)", peer.ID, err)
}

// Request & RPC Dispatching
func (d *MyDelegate) OnRequest(ctx context.Context, peer *rpc.Peer, method string, params []byte) (any, error) {
    switch method {
    case "ping":
        return "pong", nil
    case "add":
        return map[string]int{"result": 42}, nil
    default:
        // Returns standard MethodNotFound (-32601) JSON-RPC error
        return d.DefaultNexDelegate.OnRequest(ctx, peer, method, params)
    }
}

// One-way notification handler
func (d *MyDelegate) OnNotification(ctx context.Context, peer *rpc.Peer, method string, params []byte) {
    log.Printf("Notification received [%s]: %s", method, string(params))
}

2. Authentication & Session Management

DefaultNexDelegate provides built-in handling for the "auth" RPC method using token-based sessions:

1. Injecting an Authenticator: Inject a custom UserAuthenticator into your delegate or Node. If omitted, a DummyAuthenticator (georg/secret) is used as a fallback.

   myDelegate := &MyDelegate{}
   myDelegate.SetAuthenticator(myUserStore) // Implements rpc.UserAuthenticator
   
   
   

2. Protecting RPC Methods: Secure your business endpoints using peer.IsAuthorized() and retrieve identity via peer.Username():

func (d *MyDelegate) OnRequest(ctx context.Context, peer *rpc.Peer, method string, params []byte) (any, error) {
    switch method {
    case "protected.action":
        if !peer.IsAuthorized() {
            return nil, &rpc.JsonRPCerror{Code: rpc.UnAuthorized, Message: rpc.StdError[rpc.UnAuthorized]}
        }
        log.Printf("Action executed by %s", peer.Username())
        return "ok", nil
    default:
        return d.DefaultNexDelegate.OnRequest(ctx, peer, method, params)
    }
}

3. Minimal Server (Node)

Initialize the server using rpc.Options. The options are automatically unpacked into internal Node fields upon creation.

package main

import (
    "log"
    "time"

    "codeberg.org/tiny-frameworks/nexutils/logger"
    "codeberg.org/tiny-frameworks/nexutils/p2p/rpc"
)

func main() {
    _ = logger.SetupLogging(nil)

    // Create node with custom delegate
    node := rpc.NewNode(rpc.Options{
        Addr:              ":8080",
        HeartbeatInterval: 10 * time.Second,
        Delegate:          &MyDelegate{},
    })

    // Start server (blocking)
    if err := node.Start(); err != nil {
        log.Fatalf("Node crashed: %v", err)
    }
}

4. Executable Example

A a complete, runnable end-to-end example (main.go) demonstrating two nodes (Node A as Server and Node B as Client) interacting via custom NexDelegate implementations:

go run main.go

5. Managed Client with Auto-Reconnect (Resilient)

For long-lived client connections that must autonomously survive network outages and remote server restarts:

package main

import (
    "context"
    "time"

    "codeberg.org/tiny-frameworks/nexutils/p2p/rpc"
)

func main() {
    node := rpc.NewNode(rpc.Options{
        HeartbeatInterval: 5 * time.Second,
        Delegate:          &MyDelegate{},
    })
    go node.Start()
    defer node.Stop()

    // Resilient client with auto-reconnect
    client := node.ConnectWithAutoReconnect("ws://127.0.0.1:8080/ws", rpc.ReconnectConfig{
        InitialInterval: 1 * time.Second,
        MaxInterval:     10 * time.Second,
        MaxRetries:      10,
    })
    defer client.Close()

    //Set credentials for automated auth & token-based session restoration on reconnects
    client.SetCredentials("admin", "secret")

    // Synchronous call with explicit timeout
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()

    var response string
    if err := client.Call(ctx, "ping", nil, &response); err != nil {
        // Error handling
        return
    }
}

6. Direct Peer Connection (Static)

For controlled ad-hoc connections without background management:

peer, err := node.ConnectToPeer("ws://127.0.0.1:8080/ws")
if err != nil {
    // Connection establishment error
}

ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()

var result string
if err := peer.Call(ctx, "add", nil, &result); err != nil {
    // Static RPC call failed
}

Configuration Options (Options)

Running Tests

An integration test suite is included in the package directory:

go test -v ./...

Organizational & Standards

• Copyright: © 2026 Georg Hagn.
• Namespace: codeberg.org/tiny-frameworks/nexutils/p2p/rpc
• License: Apache License, Version 2.0.

GSF-nexutils/p2p is an independent open-source project and is not affiliated with any corporation of a similar name.

Contact

If you have questions or feedback, feel free to reach out:

📧 georghagn [at] tiny-frameworks.io