arpc

ARPC - More Effective Network Communication

Contents

• ARPC - More Effective Network Communication
  • Contents
  • Features
  • Performance
  • Header Layout
  • Installation
  • Quick Start
  • API Examples
    • Register Routers
    • Register Service Methods
    • Router Middleware
    • Coder Middleware
    • Client Call, CallAsync, Notify
    • Singleflight (Call De-duplication)
    • Server Call, CallAsync, Notify
    • Broadcast - Notify
    • Async Response
    • Handle New Connection
    • Handle Disconnected
    • Handle Client's send queue overstock
    • Custom Net Protocol
    • Custom Codec
    • Custom Logger
    • Custom operations before conn's recv and send
    • Custom arpc.Client's Reader by wrapping net.Conn
    • Custom arpc.Client's send queue capacity
  • JS Client
  • Web Chat Examples
  • Pub/Sub Examples
  • More Examples

Features

• Two-Way Calling
• Two-Way Notify
• Sync and Async Calling
• Sync and Async Response
• Singleflight Call De-duplication
• Batch Write | Writev | net.Buffers
• Broadcast
• Middleware
• Pub/Sub
• Opentracing

Performance

Here are some thirdparty benchmark including arpc, although these repos have provide the performance report, but I suggest you run the code yourself and get the real result, other than just believe other people's doc:

• go-rpc-framework-benchmark
• rpcx-benchmark
• kitex-benchmark

Header Layout

• LittleEndian

Installation

1. Get and install arpc

$ go get -u github.com/lesismal/arpc

1. Import in your code:

import "github.com/lesismal/arpc"

Quick Start

• start a server

package main

import "github.com/lesismal/arpc"

func main() {
	server := arpc.NewServer()

	// register router
	server.Handler.Handle("/echo", func(ctx *arpc.Context) {
		str := ""
		if err := ctx.Bind(&str); err == nil {
			ctx.Write(str)
		}
	})

	server.Run("localhost:8888")
}

• start a client

package main

import (
	"log"
	"net"
	"time"

	"github.com/lesismal/arpc"
)

func main() {
	client, err := arpc.NewClient(func() (net.Conn, error) {
		return net.DialTimeout("tcp", "localhost:8888", time.Second*3)
	})
	if err != nil {
		panic(err)
	}
	defer client.Stop()

	req := "hello"
	rsp := ""
	err = client.Call("/echo", &req, &rsp, time.Second*5)
	if err != nil {
		log.Fatalf("Call failed: %v", err)
	} else {
		log.Printf("Call Response: \"%v\"", rsp)
	}
}

API Examples

Register Routers

var handler arpc.Handler

// package
handler = arpc.DefaultHandler
// server
handler = server.Handler
// client
handler = client.Handler

// message would be default handled one by one  in the same conn reader goroutine
handler.Handle("/route", func(ctx *arpc.Context) { ... })
handler.Handle("/route2", func(ctx *arpc.Context) { ... })

// this make message handled by a new goroutine
async := true
handler.Handle("/asyncResponse", func(ctx *arpc.Context) { ... }, async)

Register Service Methods

Instead of registering each route by hand, handler.Register(m, svc) reflects over the struct value svc and registers all of its eligible methods at once, each under the "Service.Method" route name m + "." + (or just `` when m is empty). If no method is registered, Register panics.

There are two method shapes that Register recognizes:

• The first shape(a typed request/response method): an exported method with the signature func(ctx context.Context, req *Request, rsp *Response) and no return values, where req/rsp are pointers to structs.
• The second shape(an arpc.HandlerFunc method): an exported method of type func(*arpc.Context).

Register handles three cases, summarized here and shown in the examples below:

Note: Register always uses reflection while registering(a one-time cost at startup). The table above is about the call-time / per-request path: only case 1's auto-generated handler keeps using reflection on every request, so prefer case 2 or case 3 on hot paths.

In case 3 the two methods must be paired by name: the second method is named Binding. It is registered under the first method's name, and lets you write the binding/response logic yourself instead of relying on the reflection-based auto-generated handler.

…

Router Middleware

See router middleware, it's easy to implement middlewares yourself

import "github.com/lesismal/arpcex/extension/middleware/router"

var handler arpc.Handler

// package
handler = arpc.DefaultHandler
// server
handler = server.Handler
// client
handler = client.Handler

handler.Use(router.Recover())
handler.Use(router.Logger())
handler.Use(func(ctx *arpc.Context) { ... })
handler.Handle("/echo", func(ctx *arpc.Context) { ... })
handler.Use(func(ctx *arpc.Context) { ... })

Coder Middleware

• Coder Middleware is used for converting a message data to your designed format, e.g encrypt/decrypt and compress/uncompress

import "github.com/lesismal/arpcex/extension/middleware/coder/gzip"

var handler arpc.Handler

// package
handler = arpc.DefaultHandler
// server
handler = server.Handler
// client
handler = client.Handler

handler.UseCoder(gzip.New())
handler.Handle("/echo", func(ctx *arpc.Context) { ... })

Client Call, CallAsync, Notify

1. Call (Block, with timeout/context)

request := &Echo{...}
response := &Echo{}
timeout := time.Second*5
err := client.Call("/call/echo", request, response, timeout)
// ctx, cancel := context.WithTimeout(context.Background(), time.Second)
// defer cancel()
// err := client.CallWith(ctx, "/call/echo", request, response)

1. CallAsync (Nonblock, with callback and timeout/context)

request := &Echo{...}

timeout := time.Second*5
err := client.CallAsync("/call/echo", request, func(ctx *arpc.Context) {
	response := &Echo{}
	ctx.Bind(response)
	...	
}, timeout)

1. Notify (same as CallAsync with timeout/context, without callback)

data := &Notify{...}
client.Notify("/notify", data, time.Second)
// ctx, cancel := context.WithTimeout(context.Background(), time.Second)
// defer cancel()
// client.NotifyWith(ctx, "/notify", data)

Singleflight (Call De-duplication)

Enable singleflight on a router/method to de-duplicate concurrent calls: when several goroutines call the same method with the same key at the same time, only one request is actually sent to the server and all of them share its response. This cuts duplicated round-trips and server load for hot, identical requests. It applies to Call, CallContext/CallWith and CallAsync.

Only methods you register are optimized; every other method behaves exactly as before.

…

Notes:

• The leader performs the real round-trip; followers wait for it and each decodes the shared response into their own rsp. The blocking path decodes the payload only once and shares the decoded value, so the shared response should be treated as read-only.
• Every caller still honors its own timeout/context while waiting.
• Call and CallContext de-duplicate together; CallAsync de-duplicates within its own kind and each follower's callback is invoked with the shared response (or its own timeout).

Server Call, CallAsync, Notify

1. Get client and keep it in your application

var client *arpc.Client
server.Handler.Handle("/route", func(ctx *arpc.Context) {
	client = ctx.Client
	// release client
	client.OnDisconnected(func(c *arpc.Client){
		client = nil
	})
})

go func() {
	for {
		time.Sleep(time.Second)
		if client != nil {
			client.Call(...)
			client.CallAsync(...)
			client.Notify(...)
		}
	}
}()

1. Then Call/CallAsync/Notify

• See Previous

Broadcast - Notify

• for more details: server client

var mux = sync.RWMutex{}
var clientMap = make(map[*arpc.Client]struct{})

func broadcast() {
	var svr *arpc.Server = ... 
	msg := svr.NewMessage(arpc.CmdNotify, "/broadcast", fmt.Sprintf("broadcast msg %d", i))
	mux.RLock()
	for client := range clientMap {
		client.PushMsg(msg, arpc.TimeZero)
	}
	mux.RUnlock()
}

Async Response

var handl