panrpc

Start by creating a new Go module for the tutorial and installing github.com/pojntfx/panrpc/go:

$ mkdir -p panrpc-tutorial-go
$ cd panrpc-tutorial-go
$ go mod init panrpc-tutorial-go
$ go get github.com/pojntfx/panrpc/go/...@latest

The Go version of panrpc supports many transports. While common ones are TCP, WebSockets, UNIX sockets or WebRTC, anything that directly implements or can be adapted to a io.ReadWriter can be used with the panrpc LinkStream API. If you want to use a message broker like Valkey/Redis or NATS as the transport, or need more control over the wire protocol, you can use the LinkMessage API instead. For this tutorial, we'll be using WebSockets as the transport through the nhooyr.io/websocket library, which you can install like so:

$ go get nhooyr.io/websocket@latest

In addition to supporting many transports, the Go version of panrpc also supports different serializers. Common ones are JSON and CBOR, but similarly to transports anything that implements or can be adapted to a io.ReadWriter stream can be used. For this tutorial, we'll be using JSON as the serializer through the encoding/json Go standard library.

To start with implementing the coffee machine server, create a new file cmd/coffee-machine/main.go and define a basic struct with a BrewCoffee method. This method simulates brewing coffee by validating the coffee variant, checking if there is enough water available to brew the coffee, sleeping for five seconds, and returning the new water level to the remote control:

// cmd/coffee-machine/main.go

package main

import (
    "context"
    "errors"
    "log"
    "slices"
    "time"
)

type coffeeMachine struct {
    supportedVariants []string
    waterLevel        int
}

func (s *coffeeMachine) BrewCoffee(
    ctx context.Context,
    variant string,
    size int,
) (int, error) {
    if !slices.Contains(s.supportedVariants, variant) {
        return 0, errors.New("unsupported variant")
    }

    if s.waterLevel-size < 0 {
        return 0, errors.New("not enough water")
    }

    log.Println("Brewing coffee variant", variant, "in size", size, "ml")

    time.Sleep(time.Second * 5)

    s.waterLevel -= size

    return s.waterLevel, nil
}

The following limitations on which methods can be exposed as RPCs exist:

• Methods must have context.Context as their first argument
• Methods can not have variadic arguments
• Methods must return either an error or a single value and an error
• Methods must be public (private methods won't be callable as RPCs, but stay callable as regular methods)

To start turning the BrewCoffee method into an RPC, create an instance of the struct and pass it to a panrpc Registry like so:

// cmd/coffee-machine/main.go

import "github.com/pojntfx/panrpc/go/pkg/rpc"

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    service := &coffeeMachine{
        supportedVariants: []string{"latte", "americano"},
        waterLevel:        1000,
    }

    var clients atomic.Int64
    registry := rpc.NewRegistry[struct{}, json.RawMessage](
        service,

        &rpc.RegistryHooks{
            OnClientConnect: func(remoteID string) {
                log.Printf("%v remote controls connected", clients.Add(1))
            },
            OnClientDisconnect: func(remoteID string) {
                log.Printf("%v remote controls connected", clients.Add(-1))
            },
        },
    )
}

Now that we have a registry that provides our coffee machine's RPCs, we can link it to our transport (WebSockets) and serializer of choice (JSON). This requires a bit of boilerplate to upgrade from HTTP to WebSockets, so feel free to copy-and-paste this, or take a look at the examples to check out how you can set up a different transport (TCP, WebSockets, UNIX sockets etc.) and serializer (JSON, CBOR etc.) instead:

// cmd/coffee-machine/main.go

import (
    "encoding/json"
    "net"
    "net/http"

    "github.com/pojntfx/panrpc/go/pkg/rpc"
    "nhooyr.io/websocket"
)

func main() {
  // ...

  // Create TCP listener
    lis, err := net.Listen("tcp", "127.0.0.1:1337")
    if err != nil {
        panic(err)
    }
    defer lis.Close()

    log.Println("Listening on", lis.Addr())

    // Create HTTP server from TCP listener
    if err := http.Serve(lis, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        defer func() {
            if err := recover(); err != nil {
                w.WriteHeader(http.StatusInternalServerError)

                log.Printf("Remote control disconnected with error: %v", err)
            }
        }()

        // Upgrade from HTTP to WebSockets
        switch r.Method {
        case http.MethodGet:
            c, err := websocket.Accept(w, r, &websocket.AcceptOptions{
                OriginPatterns: []string{"*"},
            })
            if err != nil {
                panic(err)
            }

            pings := time.NewTicker(time.Second / 2)
            defer pings.Stop()

            errs := make(chan error)
            go func() {
                for range pings.C {
                    if err := c.Ping(ctx); err != nil {
                        errs <- err

                        return
                    }
                }
            }()

            conn := websocket.NetConn(ctx, c, websocket.MessageText)
            defer conn.Close()

            // Set up the streaming JSON encoder and decoder
            encoder := json.NewEncoder(conn)
            decoder := json.NewDecoder(conn)

            go func() {
                if err := registry.LinkStream(
          r.Context(),

                    func(v rpc.Message[json.RawMessage]) error {
                        return encoder.Encode(v)
                    },
                    func(v *rpc.Message[json.RawMessage]) error {
                        return decoder.Decode(v)
                    },

                    func(v any) (json.RawMessage, error) {
                        b, err := json.Marshal(v)
                        if err != nil {
                            return nil, err
                        }

                        return json.RawMessage(b), nil
                    },
                    func(data json.RawMessage, v any) error {
                        return json.Unmarshal([]byte(data), v)
                    },
                ); err != nil {
                    errs <- err

                    return
                }
            }()

            if err := <-errs; err != nil {
                panic(err)
            }
        default:
            w.WriteHeader(http.StatusMethodNotAllowed)
        }
    })); err != nil {
        panic(err)
    }
}

Congratulations! You've created your first panrpc server. You can start it from your terminal like so:

$ go run ./cmd/coffee-machine/main.go

You should now see the following in your terminal, which means that the server is available on localhost:1337:

Listening on localhost:1337

To start with implementing the remote control, create a new file cmd/remote-control/main.go and define a basic struct with a placeholder method that mirrors the BrewCoffee RPC:

// cmd/remote-control/main.go

package main

import "context"

type coffeeMachine struct {
    BrewCoffee func(
        ctx context.Context,
        variant string,
        size int,
    ) (int, error)
}

In order to make the BrewCoffee placeholder method do RPC calls, create an instance of the struct and pass it to a panrpc Registry like so:

// cmd/remote-control/main.go

import "github.com/pojntfx/panrpc/go/pkg/rpc"

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    var clients atomic.Int64
    registry := rpc.NewRegistry[coffeeMachine, json.RawMessage](
        &struct{}{},

        &rpc.RegistryHooks{
            OnClientConnect: func(remoteID string) {
                log.Printf("%v coffee machines connected", clients.Add(1))
            },
            OnClientDisconnect: func(remoteID string) {
                log.Printf("%v coffee machines connected", clients.Add(-1))
            },
        },
    )
}

Now that we have a registry that turns the remote control's placeholder methods into RPC calls, we can link it to our transport (WebSockets) and serializer of choice (JSON). Once again, this requires a bit of boilerplate to connect to the WebSocket, so feel free to copy-and-paste this, or take a look at the examples to check out how you can set up a different transport (TCP, WebSockets, UNIX sockets etc.) and serializer (JSON, CBOR etc.) instead:

// cmd/remote-control/main.go

import (
    "encoding/json"

    "github.com/pojntfx/panrpc/go/pkg/rpc"
    "nhooyr.io/websocket"
)

func main() {
  // ...

 // Connect to WebSocket server
    c, _, err := websocket.Dial(ctx, "ws://127.0.0.1:1337", nil)
    if err != nil {
        panic(err)
    }

    conn := websocket.NetConn(ctx, c, websocket.MessageText)
    defer conn.Close()

    log.Println("Connected to localhost:1337")

    // Set up the streaming JSON encoder and decoder
    encoder := json.NewEncoder(conn)
    decoder := json.NewDecoder(conn)

    if err := registry.LinkStream(
    ctx,

        func(v rpc.Message[json.RawMessage]) error {
            return encoder.Encode(v)
        },
        func(v *rpc.Message[json.RawMessage]) error {
            return decoder.Decode(v)
        },

        func(v any) (json.RawMessage, error) {
            b, err := json.Marshal(v)
            if err != nil {
                return nil, err
            }

            return json.RawMessage(b), nil
        },
        func(data json.RawMessage, v any) error {
            return json.Unmarshal([]byte(data), v)
        },
    ); err != nil {
        panic(err)
    }
}

Cheers! You've created your first panrpc client. You can start it from your terminal like so:

$ go run ./cmd/remote-control/main.go

You should now see the following in your terminal, which means that the client has connected to the panrpc server at localhost:1337:

Connected to localhost:1337
1 coffee machines connected

Similarly so, the coffee machine server should output the following:

1 remote controls connected

To achieve this, we can call this RPC transparently from the remote control by accessing the connected coffee machine(s) with registry.ForRemotes, and we can handle errors by checking with if err := ..., err != nil { ... } just like if we were making a local function call:

// cmd/remote-control/main.go

import (
    "bufio"
    "log"
    "os"
)

func main() {
  // ...

  go func() {
        log.Println(`Enter one of the following numbers followed by <ENTER> to brew a coffee:

- 1: Brew small Cafè Latte
- 2: Brew large Cafè Latte

- 3: Brew small Americano
- 4: Brew large Americano`)

        stdin := bufio.NewReader(os.Stdin)

        for {
            line, err := stdin.ReadString('\n')
            if err != nil {
                panic(err)
            }

            if err := registry.ForRemotes(func(remoteID string, remote coffeeMachine) error {
                switch line {
                case "1\n":
                    fallthrough
                case "2\n":
                    res, err := remote.BrewCoffee(
                        ctx,
                        "latte",
                        func() int {
                            if line == "1" {
                                return 100
                            } else {
                                return 200
                            }
                        }(),
                    )
                    if err != nil {
                        log.Println("Couldn't brew Cafè Latte:", err)

                        return nil
                    }

                    log.Println("Remaining water:", res, "ml")

                case "3\n":
                    fallthrough
                case "4\n":
                    res, err := remote.BrewCoffee(
                        ctx,
                        "americano",
                        func() int {
                            if line == "1" {
                                return 100
                            } else {
                                return 200
                            }
                        }(),
                    )
                    if err != nil {
                        log.Println("Couldn't brew Americano:", err)

                        return nil
                    }

                    log.Println("Remaining water:", res, "ml")

                default:
                    log.Printf("Unknown letter %v, ignoring input", line)

                    return nil
                }

                return nil
            }); err != nil {
                panic(err)
            }
        }
    }()

  // ...
}

Note that by cancelling the context.Context that we pass in as the first argument to every RPC call, you can cancel an RPC call before it has returned, which is useful for implementing things like timeouts. If you don't cancel this context.Context like we do in this example, the RPC call will simply block until it returns.

Now we can restart the remote control like so:

$ go run ./cmd/remote-control/main.go

After which you should see the following output:

Enter one of the following numbers followed by <ENTER> to brew a coffee:

- 1: Brew small Cafè Latte
- 2: Brew large Cafè Latte

- 3: Brew small Americano
- 4: Brew large Americano
1 coffee machines connected
Connected to localhost:1337

It is now possible to brew a coffee by pressing a number and ENTER. Once the RPC has been called, the coffee machine should print something like the following:

Brewing coffee variant latte in size 100 ml

And after the coffee has been brewed, the remote control should return the remaining water level like so:

Remaining water: 900 ml

Enjoy your (virtual) coffee! You've successfully called an RPC provided by a server from the client. Feel free to try out the other supported variants and sizes until there is no more water remaining.

To get started, we can once again create a basic struct on the client with a method SetCoffeeMachineBrewing, which will print the state of the coffee machine to the remote control's terminal:

// cmd/remote-control/main.go

type remoteControl struct{}

func (s *remoteControl) SetCoffeeMachineBrewing(ctx context.Context, brewing bool) error {
    if brewing {
        log.Println("Coffee machine is now brewing")
    } else {
        log.Println("Coffee machine has stopped brewing")
    }

    return nil
}

To start turning this new SetCoffeeMachineBrewing method into an RPC that server can call, create an instance of the struct and pass it to the client's registry like so:

// cmd/remote-control/main.go

func main() {
  // ...

  registry := rpc.NewRegistry[coffeeMachine, json.RawMessage](
        &remoteControl{},

        &rpc.RegistryHooks{
            OnClientConnect: func(remoteID string) {
                log.Printf("%v coffee machines connected", clients.Add(1))
            },
            OnClientDisconnect: func(remoteID string) {
                log.Printf("%v coffee machines connected", clients.Add(-1))
            },
        },
    )

  // ...
}

The remote control/client now exposes the SetCoffeeMachineBrewing RPC, and we can start enabling the coffee machine/server to call it by defining a basic struct with a method that mirrors the RPC, just like we did before on the remote control for BrewCoffee:

// cmd/coffee-machine/main.go

type remoteControl struct {
    SetCoffeeMachineBrewing func(ctx context.Context, brewing bool) error
}

In order to make the SetCoffeeMachineBrewing placeholder method do RPC calls, create an instance of the struct and pass it to the server's registry like so:

// cmd/coffee-machine/main.go

func main() {
  // ...

    registry := rpc.NewRegistry[remoteControl, json.RawMessage](
        service,

        &rpc.RegistryHooks{
            OnClientConnect: func(remoteID string) {
                log.Printf("%v remote controls connected", clients.Add(1))
            },
            OnClientDisconnect: func(remoteID string) {
                log.Printf("%v remote controls connected", clients.Add(-1))
            },
        },
    )

  // ...
}

The coffee machine/server and the remote control/client now both know of the new SetCoffeeMachineBrewing RPC, but the server doesn't call it yet. To fix this, we can call this RPC transparently from the coffee machine by accessing the connected remote control(s) with registry.ForRemotes just like we did before in the remote control, and we can handle errors by checking with if err := ..., err != nil { ... } just like if we were making a local function call. We'll also use the first argument to the RPC, ctx, in conjunction with rpc.GetRemoteID to get the ID of the remote control/client that is calling BrewCoffee, so that we don't call SetCoffeeMachineBrewing on the remote control/client that is calling BrewCoffee itself:

// cmd/remote-control/main.go

type coffeeMachine struct {
    supportedVariants []string
    waterLevel        int

    ForRemotes func(
        cb func(remoteID string, remote remoteControl) error,
    ) error
}

func (s *coffeeMachine) BrewCoffee(
    ctx context.Context,
    variant string,
    size int,
) (int, error) {
  // Get the ID of the remote control that's calling `BrewCoffee`
    targetID := rpc.GetRemoteID(ctx)

  // Notify connected remote controls that coffee is no longer brewing
    defer s.ForRemotes(func(remoteID string, remote remoteControl) error {
    // Don't call `SetCoffeeMachineBrewing` if it's the remote control that's calling `BrewCoffee`
        if remoteID == targetID {
            return nil
        }

        return remote.SetCoffeeMachineBrewing(ctx, false)
    })

  // Notify connected remote controls that coffee is brewing
    if err := s.ForRemotes(func(remoteID string, remote remoteControl) error {
    // Don't call `SetCoffeeMachineBrewing` if it's the remote control that's calling `BrewCoffee`
        if remoteID == targetID {
            return nil
        }

        return remote.SetCoffeeMachineBrewing(ctx, true)
    }); err != nil {
        return 0, err
    }

    if !slices.Contains(s.supportedVariants, variant) {
        return 0, errors.New("unsupported variant")
    }

    if s.waterLevel-size < 0 {
        return 0, errors.New("not enough water")
    }

    log.Println("Brewing coffee variant", variant, "in size", size, "ml")

    time.Sleep(time.Second * 5)

    s.waterLevel -= size

    return s.waterLevel, nil
}

Note that we've added the forRemotes field to the coffee machine/server; we can get the implementation for it from the registry like so:

// cmd/coffee-machine/main.go

func main{
  service := // ...

  registry := // ...
  service.forRemotes = registry.ForRemotes;
}

Now that we've added support for this RPC to the coffee machine/server, we can restart it like so:

$ go run ./cmd/coffee-machine/main.go

To test if it works, connect two remote controls/clients to it like so:

$ go run ./cmd/remote-control/main.go
# In another terminal
$ go run ./cmd/remote-control/main.go

You can now request the coffee machine to brew a coffee on either of the remote controls by pressing a number and ENTER. Once the RPC has been called, the coffee machine should print something like the following again:

Brewing coffee variant latte in size 100 ml

And after the coffee has been brewed, the remote control that you've chosen to brew the coffee with should once again return the remaining water level like so:

Remaining water: 900 ml

The other connected remote controls will be notified that the coffee machine is brewing, and then once it has finished brewing:

Coffee machine is now brewing
Coffee machine has stopped brewing

Enjoy your distributed coffee machine! You've successfully called an RPC provided by a client from the server to implement multicast notifications, something that usually is quite complex to do with RPC systems.

First, we'll add a onProgress callback to the coffee machine's BrewCoffee implementation, which we then call incrementally during the brewing process:

// cmd/coffee-machine/main.go

func (s *coffeeMachine) BrewCoffee(
    ctx context.Context,
    variant string,
    size int,
    onProgress func(ctx context.Context, percentage int) error, // This is new
) (int, error) {
    // ...

    // Report 0% brewing process
    if err := onProgress(ctx, 0); err != nil {
        return 0, err
    }

    // Report 25% brewing process
    time.Sleep(500 * time.Millisecond)
    if err := onProgress(ctx, 25); err != nil {
        return 0, err
    }

    // Report 50% brewing process
    time.Sleep(500 * time.Millisecond)
    if err := onProgress(ctx, 50); err != nil {
        return 0, err
    }

    // Report 75% brewing process
    time.Sleep(500 * time.Millisecond)
    if err := onProgress(ctx, 75); err != nil {
        return 0, err
    }

    // Report 100% brewing process
    time.Sleep(500 * time.Millisecond)
    if err := onProgress(ctx, 100); err != nil {
        return 0, err
    }

    // ..

    return s.waterLevel, nil
}

In the remote control, we'll also extend the struct with the BrewCoffee placeholder method with this new RPC argument:

// cmd/remote-control/main.go

type coffeeMachine struct {
    BrewCoffee func(
        ctx context.Context,
        variant string,
        size int,
        onProgress func(ctx context.Context, percentage int) error, // This is new
    ) (int, error)
}

And finally, where we call the BrewCoffee RPC in the remote control/client, we can pass in the implementation of this closure:

// cmd/remote-control/main.go

go func() {
    // ...
    for {
        // ...
        if err := registry.ForRemotes(func(remoteID string, remote coffeeMachine) error {
            switch line {
            case "1\n":
                fallthrough
            case "2\n":
                res, err := remote.BrewCoffee(
                    ctx,
                    "latte",
                    func() int {
                        if line == "1" {
                            return 100
                        } else {
                            return 200
                        }
                    }(),
                    func(ctx context.Context, percentage int) error {
                        log.Printf(`Brewing Cafè Latte ... %v%% done`, percentage) // This is new

                        return nil
                    },
                )

                // ...

            case "3\n":
                fallthrough
            case "4\n":
                res, err := remote.BrewCoffee(
                    ctx,
                    "americano",
                    func() int {
                        if line == "1" {
                            return 100
                        } else {
                            return 200
                        }
                    }(),
                    func(ctx context.Context, percentage int) error {
                        log.Printf(`Brewing Americano ... %v%% done`, percentage) // This is new

                        return nil
                    },
                )

                // ...

            return nil
        }); err != nil {
            panic(err)
        }
    }
}()

Now we can restart the coffee machine/server again like so:

$ go run ./cmd/coffee-machine/main.go

And connect the remote control/client to it again like so:

$ go run ./cmd/remote-control/main.go

You can now request the coffee machine to brew a coffee by pressing a number and ENTER. Once the RPC has been called, the coffee machine should print something like the following again:

Brewing coffee variant latte in size 100 ml

And the remote control will print the progress as reported by the coffee machine to the terminal, before once again returning the remaining water level like so:

Brewing Cafè Latte ... 0% done
Brewing Cafè Latte ... 25% done
Brewing Cafè Latte ... 50% done
Brewing Cafè Latte ... 75% done
Brewing Cafè Latte ... 100% done
Remaining water: 900 ml

Enjoy your live coffee brewing progress! You've successfully implemented incremental coffee brewing progress reports by using panrpc's closure support, something that is usually quite tricky to do with RPC frameworks.

To define a nested RPC, simply add another struct as a public property to your existing coffee machine/server or remote control/client. In this new stuct, you can define RPCs as methods, just like with top-level RPCs. To call them, follow the same concept: Define placeholder methods in the new struct and add it as a public property to your coffee machine/server or remote control/client.

In this example, we'll add a tea brewer extension to our coffee machine/server and remote control/client. This extension will allow us to list available tea variants by calling the Extension.GetVariants RPC. For brevity, we won't implement all the tea brewing functions. To add the tea brewer extension to the coffee machine/server, first create a new TeaBrewer struct, similar to the CoffeeMachine struct. Then, add the GetVariants method to this class and instantiate it with the supported variants:

// cmd/coffee-machine/main.go

type teaBrewer struct {
    supportedVariants []string
}

func (s *teaBrewer) GetVariants(ctx context.Context) ([]string, error) {
    return s.supportedVariants, nil
}

To add these nested RPCs to the main CoffeeMachine struct, we'll include them as a public property in the struct. To keep the coffee machine flexible and avoid depending on the TeaBrewer extension, we'll name the property Extension and make it generic in CoffeeMachine. This way, we can easily replace the tea brewer with another extension in the future, such as a full-featured tea brewer:

// cmd/coffee-machine/main.go

type coffeeMachine[E any] struct {
    Extension E

  // ...
}

Finally, for the coffee machine/server, create an instance of our extension and pass it to the CoffeeMachine when initializing it. This is also where you provide the list of supported tea variants:

// cmd/coffee-machine/main.go

// ...
service := &coffeeMachine[*teaBrewer]{
        Extension: &teaBrewer{
            supportedVariants: []string{"darjeeling", "chai", "earlgrey"},
        },

        supportedVariants: []string{"latte", "americano"},
        waterLevel:        1000,
    }
// ...

For the remote control/client, it's a very similar process. First, we define the new TeaBrewer struct with the GetVariants placeholder method:

// cmd/remote-control/main.go

type teaBrewer struct {
    GetVariants func(ctx context.Context) ([]string, error)
}

Then we'll add the nested RPCs to the main CoffeeMachine struct as a public instance property, and we'll use generics again to keep everything extensible:

// cmd/remote-control/main.go

type coffeeMachine[E any] struct {
    Extension E

    // ...
}

After this, we need to create an instance of our extension and pass it to CoffeeMachine when we create it:

// cmd/remote-control/main.go

registry := rpc.NewRegistry[coffeeMachine[teaBrewer], json.RawMessage](
        &remoteControl{},

        // ...
    )

And finally, we add another switch case to the remote control/client so that we can call Extension.GetVariants:

// cmd/remote-control/main.go

    go func() {
        log.Println(`Enter one of the following numbers followed by <ENTER> to brew a coffee:

- 1: Brew small Cafè Latte
- 2: Brew large Cafè Latte

- 3: Brew small Americano
- 4: Brew large Americano

Or enter 5 to list available tea variants.`)

  // ...

  if err := registry.ForRemotes(func(remoteID string, remote coffeeMachine[teaBrewer]) error {
        switch line {
      // ..
      case "5\n":
                res, err := remote.Extension.GetVariants(ctx)
                if err != nil {
                    log.Println("Couldn't list available tea variants:", err)

                    return nil
                }

                log.Println("Available tea variants:", res)

      default:
      // ..
        }
    }
}()

Now we can restart the coffee machine/server again like so:

$ go run ./cmd/coffee-machine/main.go

And connect the remote control/client to it again like so:

$ go run ./cmd/remote-control/main.go

You can now request the coffee machine to list the available tea variants by pressing 5 and ENTER. Once the RPC has been called, the remote control should print something like the following:

Available tea variants: [ "darjeeling", "chai", "earlgrey" ]

🚀 That's it! You've successfully built a virtual coffee machine with support for brewing coffee, notifications when coffee is being brewed, and incremental coffee brewing progress reports. You've also made it easily extensible by using nested RPCs. We can't wait to see what you're going to build next with panrpc! Be sure to take a look at the reference and examples for more information, or check out the complete sources for the coffee machine server and coffee machine client/remote control for a recap.

Start by creating a new npm module for the tutorial and installing @pojntfx/panrpc:

$ mkdir -p panrpc-tutorial-typescript
$ cd panrpc-tutorial-typescript
$ npm init -y
$ npm install @pojntfx/panrpc

The TypeScript version of panrpc supports many transports. While common ones are TCP, WebSockets, UNIX sockets or WebRTC, anything that directly implements or can be adapted to a WHATWG stream can be used with the panrpc linkStream API. If you want to use a message broker like Valkey/Redis or NATS as the transport, or need more control over the wire protocol, you can use the linkMessage API instead. For this tutorial, we'll be using WebSockets as the transport through the ws library, which you can install like so:

$ npm install ws

In addition to supporting many transports, the TypeScript version of panrpc also supports different serializers. Common ones are JSON and CBOR, but similarly to transports anything that implements or can be adapted to a WHATWG stream can be used. For this tutorial, we'll be using JSON as the serializer through the @streamparser/json-whatwg library, which you can install like so:

$ npm install @streamparser/json-whatwg

To start with implementing the coffee machine server, create a new file coffee-machine.ts and define a basic class with a BrewCoffee method. This method simulates brewing coffee by validating the coffee variant, checking if there is enough water available to brew the coffee, sleeping for five seconds, and returning the new water level to the remote control:

// coffee-machine.ts

import { ILocalContext } from "@pojntfx/panrpc";

class CoffeeMachine {
  #waterLevel: number;

  constructor(private supportedVariants: string[], waterLevel: number) {
    this.#waterLevel = waterLevel;

    this.BrewCoffee = this.BrewCoffee.bind(this);
  }

  async BrewCoffee(
    ctx: ILocalContext,
    variant: string,
    size: number
  ): Promise<number> {
    if (!this.supportedVariants.includes(variant)) {
      throw new Error("unsupported variant");
    }

    if (this.#waterLevel - size < 0) {
      throw new Error("not enough water");
    }

    console.log("Brewing coffee variant", variant, "in size", size, "ml");

    await new Promise((r) => {
      setTimeout(r, 5000);
    });

    this.#waterLevel -= size;

    return this.#waterLevel;
  }
}

The following limitations on which methods can be exposed as RPCs exist:

• Methods must have ILocalContext as their first argument
• Methods can not have variadic arguments
• Methods must be public (private methods - those that start with #, see private properties - won't be callable as RPCs, but stay callable as regular methods)

To start turning the BrewCoffee method into an RPC, create an instance of the class and pass it to a panrpc Registry like so:

// coffee-machine.ts

import { Registry } from "@pojntfx/panrpc";

const service = new CoffeeMachine(["latte", "americano"], 1000);

let clients = 0;

const registry = new Registry(
  service,
  new (class {})(),

  {
    onClientConnect: () => {
      clients++;

      console.log(clients, "remote controls connected");
    },
    onClientDisconnect: () => {
      clients--;

      console.log(clients, "remote controls connected");
    },
  }
);

Now that we have a registry that provides our coffee machine's RPCs, we can link it to our transport (WebSockets) and serializer of choice (JSON). This requires a bit of boilerplate since the ws library doesn't provide WHATWG streams directly yet, so feel free to copy-and-paste this, or take a look at the examples to check out how you can set up a different transport (TCP, WebSockets, UNIX sockets etc.) and serializer (JSON, CBOR etc.) instead:

// coffee-machine.ts

import { JSONParser } from "@streamparser/json-whatwg";
import { WebSocketServer } from "ws";

// Create WebSocket server
const server = new WebSocketServer({
  host: "127.0.0.1",
  port: 1337,
});

server.on("connection", (socket) => {
  socket.addEventListener("error", (e) => {
    console.error("Remote control disconnected with error:", e);
  });

  const linkSignal = new AbortController();

  // Set up streaming JSON encoder
  const encoder = new WritableStream({
    write(chunk) {
      socket.send(JSON.stringify(chunk));
    },
  });

  // Set up streaming JSON decoder
  const parser = new JSONParser({
    paths: ["$"],
    separator: "",
  });
  const parserWriter = parser.writable.getWriter();
  const parserReader = parser.readable.getReader();
  const decoder = new ReadableStream({
    start(controller) {
      parserReader
        .read()
        .then(async function process({ done, value }) {
          if (done) {
            controller.close();

            return;
          }

          controller.enqueue(value?.value);

          parserReader
            .read()
            .then(process)
            .catch((e) => controller.error(e));
        })
        .catch((e) => controller.error(e));
    },
  });
  socket.addEventListener("message", (m) =>
    parserWriter.write(m.data as string)
  );
  socket.addEventListener("close", () => {
    parserReader.cancel();
    parserWriter.abort();
    linkSignal.abort();
  });

  registry.linkStream(
    linkSignal.signal,

    encoder,
    decoder,

    (v) => v,
    (v) => v
  );
});

console.log("Listening on localhost:1337");

Congratulations! You've created your first panrpc server. You can start it from your terminal like so:

$ npx tsx coffee-machine.ts

You should now see the following in your terminal, which means that the server is available on localhost:1337:

Listening on localhost:1337

To start with implementing the remote control, create a new file remote-control.ts and define a basic class with a placeholder method that mirrors the BrewCoffee RPC:

// remote-control.ts

import { IRemoteContext } from "@pojntfx/panrpc";

class CoffeeMachine {
  async BrewCoffee(
    ctx: IRemoteContext,
    variant: string,
    size: number
  ): Promise<number> {
    return 0;
  }
}

Placeholder methods must have IRemoteContext instead of ILocalContext as their first argument.

In order to make the BrewCoffee placeholder method do RPC calls, create an instance of the class and pass it to a panrpc Registry like so:

// remote-control.ts

import { Registry } from "@pojntfx/panrpc";

let clients = 0;

const registry = new Registry(
  new (class {})(),
  new CoffeeMachine(),

  {
    onClientConnect: () => {
      clients++;

      console.log(clients, "coffee machines connected");
    },
    onClientDisconnect: () => {
      clients--;

      console.log(clients, "coffee machines connected");
    },
  }
);

Now that we have a registry that turns the remote control's placeholder methods into RPC calls, we can link it to our transport (WebSockets) and serializer of choice (JSON). Once again, this requires a bit of boilerplate since the ws library doesn't provide WHATWG streams directly yet, so feel free to copy-and-paste this, or take a look at the examples to check out how you can set up a different transport (TCP, WebSockets, UNIX sockets etc.) and serializer (JSON, CBOR etc.) instead:

// remote-control.ts

import { JSONParser } from "@streamparser/json-whatwg";
import { WebSocket } from "ws";

// Connect to WebSocket server
const socket = new WebSocket("ws://127.0.0.1:1337");

socket.addEventListener("error", (e) => {
  console.error("Disconnected with error:", e);

  exit(1);
});
socket.addEventListener("close", () => exit(0));

await new Promise<void>((res, rej) => {
  socket.addEventListener("open", () => res());
  socket.addEventListener("error", rej);
});

const linkSignal = new AbortController();

// Set up streaming JSON encoder
const encoder = new WritableStream({
  write(chunk) {
    socket.send(JSON.stringify(chunk));
  },
});

// Set up streaming JSON decoder
const parser = new JSONParser({
  paths: ["$"],
  separator: "",
});
const parserWriter = parser.writable.getWriter();
const parserReader = parser.readable.getReader();
const decoder = new ReadableStream({
  start(controller) {
    parserReader
      .read()
      .then(async function process({ done, value }) {
        if (done) {
          controller.close();

          return;
        }

        controller.enqueue(value?.value);

        parserReader
          .read()
          .then(process)
          .catch((e) => controller.error(e));
      })
      .catch((e) => controller.error(e));
  },
});
socket.addEventListener("message", (m) => parserWriter.write(m.data as string));
socket.addEventListener("close", () => {
  parserReader.cancel();
  parserWriter.abort();
  linkSignal.abort();
});

registry.linkStream(
  linkSignal.signal,

  encoder,
  decoder,

  (v) => v,
  (v) => v
);

console.log("Connected to localhost:1337");

Cheers! You've created your first panrpc client. You can start it from your terminal like so:

$ npx tsx remote-control.ts

You should now see the following in your terminal, which means that the client has connected to the panrpc server at localhost:1337:

Connected to localhost:1337
1 coffee machines connected

Similarly so, the coffee machine server should output the following:

1 remote controls connected

To achieve this, we can call this RPC transparently from the remote control by accessing the connected coffee machine(s) with registry.forRemotes, and we can handle errors with try catch just like if we were making a local function call:

// remote-control.ts

import { createInterface } from "readline/promises";

(async () => {
  console.log(`Enter one of the following numbers followed by <ENTER> to brew a coffee:

- 1: Brew small Cafè Latte
- 2: Brew large Cafè Latte

- 3: Brew small Americano
- 4: Brew large Americano`);

  const rl = createInterface({ input: stdin, output: stdout });

  while (true) {
    const line = await rl.question("");

    await registry.forRemotes(async (remoteID, remote) => {
      switch (line) {
        case "1":
        case "2":
          try {
            const res = await remote.BrewCoffee(
              undefined,
              "latte",
              line === "1" ? 100 : 200
            );

            console.log("Remaining water:", res, "ml");
          } catch (e) {
            console.error(`Couldn't brew Cafè Latte: ${e}`);
          }

          break;

        case "3":
        case "4":
          try {
            const res = await remote.BrewCoffee(
              undefined,
              "americano",
              line === "3" ? 100 : 200
            );

            console.log("Remaining water:", res, "ml");
          } catch (e) {
            console.error(`Couldn't brew Americano: ${e}`);
          }

          break;

        default:
          console.log(`Unknown letter ${line}, ignoring input`);
      }
    });
  }
})();

Note that by aborting the AbortSignal that we can pass in as the first argument to every RPC call, you can abort an RPC call before it has returned, which is useful for implementing things like timeouts. If you don't abort this AbortSignal, or pass in undefined like we do in this example, the RPC call will simply block until it returns.

Now we can restart the remote control like so:

$ npx tsx remote-control.ts

After which you should see the following output:

Enter one of the following numbers followed by <ENTER> to brew a coffee:

- 1: Brew small Cafè Latte
- 2: Brew large Cafè Latte

- 3: Brew small Americano
- 4: Brew large Americano
1 coffee machines connected
Connected to localhost:1337

It is now possible to brew a coffee by pressing a number and ENTER. Once the RPC has been called, the coffee machine should print something like the following:

Brewing coffee variant latte in size 100 ml

And after the coffee has been brewed, the remote control should return the remaining water level like so:

Remaining water: 900 ml

Enjoy your (virtual) coffee! You've successfully called an RPC provided by a server from the client. Feel free to try out the other supported variants and sizes until there is no more water remaining.

To get started, we can once again create a basic class on the client with a method SetCoffeeMachineBrewing, which will print the state of the coffee machine to the remote control's terminal:

// remote-control.ts

class RemoteControl {
  async SetCoffeeMachineBrewing(ctx: ILocalContext, brewing: boolean) {
    if (brewing) {
      console.log("Coffee machine is now brewing");
    } else {
      console.log("Coffee machine has stopped brewing");
    }
  }
}

To start turning this new SetCoffeeMachineBrewing method into an RPC that server can call, create an instance of the class and pass it to the client's registry like so:

// remote-control.ts

const registry = new Registry(
  new RemoteControl(), // This line is new
  new CoffeeMachine(),

  {
    onClientConnect: () => {
      clients++;

      console.log(clients, "coffee machines connected");
    },
    onClientDisconnect: () => {
      clients--;

      console.log(clients, "coffee machines connected");
    },
  }
);

The remote control/client now exposes the SetCoffeeMachineBrewing RPC, and we can start enabling the coffee machine/server to call it by defining a basic class with a method that mirrors the RPC, just like we did before on the remote control for BrewCoffee:

// coffee-machine.ts

class RemoteControl {
  async SetCoffeeMachineBrewing(ctx: IRemoteContext, brewing: boolean) {}
}

In order to make the SetCoffeeMachineBrewing placeholder method do RPC calls, create an instance of the class and pass it to the server's registry like so:

// coffee-machine.ts

const registry = new Registry(
  service,
  new RemoteControl(), // This line is new

  {
    onClientConnect: () => {
      clients++;

      console.log(clients, "remote controls connected");
    },
    onClientDisconnect: () => {
      clients--;

      console.log(clients, "remote controls connected");
    },
  }
);

The coffee machine/server and the remote control/client now both know of the new SetCoffeeMachineBrewing RPC, but the server doesn't call it yet. To fix this, we can call this RPC transparently from the coffee machine by accessing the connected remote control(s) with registry.forRemotes just like we did before in the remote control, and we can handle errors with try catch just like if we were making a local function call. We'll also use the first argument to the RPC, ILocalContext, to get the ID of the remote control/client that is calling BrewCoffee, so that we don't call SetCoffeeMachineBrewing on the remote control/client that is calling BrewCoffee itself:

// coffee-machine.ts

class CoffeeMachine {
  public forRemotes?: (
    cb: (remoteID: string, remote: RemoteControl) => Promise<void>
  ) => Promise<void>;

  // ...

  async BrewCoffee(
    ctx: ILocalContext,
    variant: string,
    size: number
  ): Promise<number> {
    // Get the ID of the remote control that's calling `BrewCoffee`
    const { remoteID: targetID } = ctx;

    try {
      // Notify connected remote controls that coffee is brewing
      await this.forRemotes?.(async (remoteID, remote) => {
        // Don't call `SetCoffeeMachineBrewing` if it's the remote control that's calling `BrewCoffee`
        if (remoteID === targetID) {
          return;
        }

        await remote.SetCoffeeMachineBrewing(undefined, true);
      });

      if (!this.supportedVariants.includes(variant)) {
        throw new Error("unsupported variant");
      }

      if (this.#waterLevel - size < 0) {
        throw new Error("not enough water");
      }

      console.log("Brewing coffee variant", variant, "in size", size, "ml");

      await new Promise((r) => {
        setTimeout(r, 5000);
      });
    } finally {
      // Notify connected remote controls that coffee is no longer brewing
      await this.forRemotes?.(async (remoteID, remote) => {
        // Don't call `SetCoffeeMachineBrewing` if it's the remote control that's calling `BrewCoffee`
        if (remoteID === targetID) {
          return;
        }

        await remote.SetCoffeeMachineBrewing(undefined, false);
      });
    }

    this.#waterLevel -= size;

    return this.#waterLevel;
  }
}

Note that we've added the forRemotes field to the coffee machine/server; we can get the implementation for it from the registry like so:

// coffee-machine.ts

const service = // ...

const registry = // ...

service.forRemotes = registry.forRemotes;

Now that we've added support for this RPC to the coffee machine/server, we can restart it like so:

$ npx tsx coffee-machine.ts

To test if it works, connect two remote controls/clients to it like so:

$ npx tsx remote-control.ts
# In another terminal
$ npx tsx remote-control.ts

You can now request the coffee machine to brew a coffee on either of the remote controls by pressing a number and ENTER. Once the RPC has been called, the coffee machine should print something like the following again:

Brewing coffee variant latte in size 100 ml

And after the coffee has been brewed, the remote control that you've chosen to brew the coffee with should once again return the remaining water level like so:

Remaining water: 900 ml

The other connected remote controls will be notified that the coffee machine is brewing, and then once it has finished brewing:

Coffee machine is now brewing
Coffee machine has stopped brewing

Enjoy your distributed coffee machine! You've successfully called an RPC provided by a client from the server to implement multicast notifications, something that usually is quite complex to do with RPC systems.

First, we'll add a onProgress callback to the coffee machine's BrewCoffee implementation and decorate it with [panrpc's