qrpc

README

QRPC

A simple rpc framework that works over QUIC written in Golang.

Client sample code

package main

import (
	"log"
	"time"

	"github.com/zoujiaqing/qrpc"
)

func main() {
	client := qrpc.NewClient("localhost", 4444)

	if err := client.Connect(); err != nil {
		return err
	}

	client.OnRequest(func(data []byte) []byte {
		return append([]byte("From Client Respond "), data...)
	})

	var i uint64 = 1
	for {
		data, err := client.Request([]byte("Hello"))
		if err != nil {
			log.Printf("Request error: %v", err)
			break
		}
		log.Printf("Respond(%d): %s", i, string(data))
		time.Sleep(1 * time.Second)
		i++
	}
}

Server sample code

package main

import (
	"context"

	"github.com/zoujiaqing/qrpc"
)

func main() {
	server, err := qrpc.NewServer(4444)
	if err != nil {
		return err
	}
	defer func() { _ = server.Close() }()

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

	server.Accept(ctx)
}

Run examples

examples include grpc client and server sample code

Load dependency

go mod dity

Run server for example

go run ./examples/server

Run client for example

go run ./examples/client

Documentation

Index

• type Client
  • func NewClient(addr string, port uint) *Client
  • func (c *Client) Connect() error
  • func (c *Client) Disconnect()
  • func (c *Client) GetPingValue() int
  • func (c *Client) OnRequest(handle RequestMessageHandler)
  • func (c *Client) Request(data []byte) ([]byte, error)
• type ConnectionHandler
• type IDGenerator
  • func (m *IDGenerator) Next() uint64
• type Message
  • func (m *Message) Read(r io.Reader) error
  • func (m *Message) Write(w io.Writer) error
• type MessageBroker
  • func NewMessageBroker(timeout time.Duration) *MessageBroker
  • func (b *MessageBroker) ReceiveResponse(response *Message) bool
  • func (b *MessageBroker) Send(request *Message) (*Message, error)
• type MessageHandler
• type MessagePair
• type RequestMessageHandler
• type RpcConnection
  • func NewRpcConnection(id uint64, ctx context.Context, conn quic.Connection, timeout uint) *RpcConnection
  • func (c *RpcConnection) AddMetadata(key, value string)
  • func (c *RpcConnection) Close()
  • func (c *RpcConnection) GetMetadata(key string) (string, bool)
  • func (c *RpcConnection) ID() uint64
  • func (c *RpcConnection) OnClose(handle func())
  • func (c *RpcConnection) OnRequest(handle MessageHandler)
  • func (c *RpcConnection) Ping() (int, error)
  • func (c *RpcConnection) RemoveMetadata(key string)
  • func (c *RpcConnection) Request(data []byte) ([]byte, error)
• type Server
  • func NewServer(port uint) (*Server, error)
  • func (s *Server) Accept(ctx context.Context)
  • func (s *Server) Close() error
  • func (s *Server) OnConnection(handle ConnectionHandler)

Types

type Client

type Client struct {
	Addr       string
	Port       uint
	RetryDelay time.Duration
	// contains filtered or unexported fields
}

func NewClient

func NewClient(addr string, port uint) *Client

func (*Client) Connect

func (c *Client) Connect() error

func (*Client) Disconnect

func (c *Client) Disconnect()

func (*Client) GetPingValue

func (c *Client) GetPingValue() int

func (*Client) OnRequest

func (c *Client) OnRequest(handle RequestMessageHandler)

func (*Client) Request

func (c *Client) Request(data []byte) ([]byte, error)

type ConnectionHandler

type ConnectionHandler func(*RpcConnection)

type IDGenerator

type IDGenerator struct {
	// contains filtered or unexported fields
}

IDGenerator 用于生成唯一的消息ID

func (*IDGenerator) Next

func (m *IDGenerator) Next() uint64

Next 生成下一个唯一的消息ID

type Message

type Message struct {
	ID        uint64
	Type      uint64
	IsAck     bool
	Data      []byte
	Timestamp int64
}

Message 表示一个消息

func (*Message) Read

func (m *Message) Read(r io.Reader) error

Read 从 io.Reader 中读取消息数据

func (*Message) Write

func (m *Message) Write(w io.Writer) error

Write 将消息数据写入到 io.Writer 中

type MessageBroker

type MessageBroker struct {
	// contains filtered or unexported fields
}

MessageBroker 负责管理消息配对和等待回复

func NewMessageBroker

func NewMessageBroker(timeout time.Duration) *MessageBroker

NewMessageBroker 创建一个新的消息 Broker

func (*MessageBroker) ReceiveResponse

func (b *MessageBroker) ReceiveResponse(response *Message) bool

ReceiveResponse 接收回复消息，并通知等待的请求方

func (*MessageBroker) Send

func (b *MessageBroker) Send(request *Message) (*Message, error)

Send 发送消息并等待回复，如果超时返回错误

type MessageHandler

type MessageHandler func(*RpcConnection, []byte) []byte

type MessagePair

type MessagePair struct {
	Request  *Message
	Response *Message
}

MessagePair 表示一个消息和其回复消息的配对

type RequestMessageHandler

type RequestMessageHandler func(data []byte) []byte

type RpcConnection

type RpcConnection struct {
	// contains filtered or unexported fields
}

func NewRpcConnection

func NewRpcConnection(id uint64, ctx context.Context, conn quic.Connection, timeout uint) *RpcConnection

func (*RpcConnection) AddMetadata

func (c *RpcConnection) AddMetadata(key, value string)

func (*RpcConnection) Close

func (c *RpcConnection) Close()

Close 关闭连接

func (*RpcConnection) GetMetadata

func (c *RpcConnection) GetMetadata(key string) (string, bool)

func (*RpcConnection) ID

func (c *RpcConnection) ID() uint64

func (*RpcConnection) OnClose

func (c *RpcConnection) OnClose(handle func())

func (*RpcConnection) OnRequest

func (c *RpcConnection) OnRequest(handle MessageHandler)

func (*RpcConnection) Ping

func (c *RpcConnection) Ping() (int, error)

func (*RpcConnection) RemoveMetadata

func (c *RpcConnection) RemoveMetadata(key string)

func (*RpcConnection) Request

func (c *RpcConnection) Request(data []byte) ([]byte, error)

type Server

type Server struct {
	// contains filtered or unexported fields
}

func NewServer

func NewServer(port uint) (*Server, error)

func (*Server) Accept

func (s *Server) Accept(ctx context.Context)

func (*Server) Close

func (s *Server) Close() error

func (*Server) OnConnection

func (s *Server) OnConnection(handle ConnectionHandler)