geminio

README

One connection. Bidirectional RPC, acked messaging, and stream multiplexing — behind a single net.Conn.

English | 简体中文 | 日本語 | 한국어 | Español | Français | Deutsch

---

Why Geminio?

You're building an IM server, a message queue, an API gateway, a reverse tunnel for NAT traversal, or a service-mesh sidecar. To get it right you need bidirectional RPC, reliable messaging with acks, many logical streams over one TCP connection, automatic reconnect, and all of it has to play nicely with Go's net.Conn / net.Listener.

The usual answer is: gRPC for RPC, yamux/smux for multiplexing, NATS or a custom protocol for messaging, and a tangle of glue to keep their lifecycles in sync. Geminio offers the whole bundle behind one interface.

Geminio vs. the usual suspects

	gRPC	yamux / smux	NATS	Geminio
Request / response RPC	✅	—	—	✅
Server-initiated RPC to client	⚠️ streaming only	—	—	✅
Messaging with publish/ack	—	—	✅	✅
Stream multiplexing	✅ (HTTP/2)	✅	—	✅
Drop-in net.Conn / net.Listener	—	✅	—	✅
Client-side auto-reconnect	—	—	✅	✅
Single binary, no broker	✅	✅	—	✅

"Server-initiated RPC" means the server can Call("method", ...) a handler the client registered — not just push messages on an open stream. It's the piece most "RPC libraries" don't ship.

Features

• 🔄 Bidirectional RPC — either side can register methods and call the other's.
• 📨 Acked messaging — Publish / Receive with delivery confirmation; sync and async.
• 🔀 Stream multiplexing — open any number of logical streams over one connection.
• 🔌 net.Conn / net.Listener compatible — streams drop into any code that speaks Go's net interfaces.
• 🆔 Stable peer & stream IDs — ClientID and StreamID make routing, authz, and tracing straightforward.
• 🔁 Auto-reconnect — client resumes transparently after network blips.
• ⚡ ~1.3 GB/s stream throughput on a 2016 laptop CPU (see Benchmarks).
• 🧪 Hardened — unit, integration, e2e, stress, chaos, and regression test suites.

60-second demo: push a file from server to client

go get github.com/singchia/geminio

Every Geminio stream is a net.Conn, and every End is a net.Listener. So a server-initiated file transfer is just io.Copy — no framing, no codec, no broker.

Server — accept clients, open a stream back, copy the file in.

ln, _ := server.Listen("tcp", "127.0.0.1:8080")
for {
    end, _ := ln.AcceptEnd()
    go func() {
        stream, _ := end.OpenStream()
        defer stream.Close()
        f, _ := os.Open("payload.bin")
        defer f.Close()
        io.Copy(stream, f)
    }()
}

Client — treat the End as a net.Listener, save each incoming stream.

end, _ := client.NewEnd("tcp", "127.0.0.1:8080")
defer end.Close()
for {
    conn, _ := end.Accept()
    f, _ := os.Create("received.bin")
    io.Copy(f, conn)
    f.Close()
    conn.Close()
}

The server initiates. The client listens on its own dial-out connection. io.Copy does the rest because the stream speaks net.Conn. Full runnable examples — RPC, bidirectional RPC, acked messaging, more multiplexing — in docs/USAGE.md.

What you can build

Scenario	Why Geminio fits	Example
NAT traversal / reverse tunnel	one outbound connection carries bidirectional control + many data streams	examples/traversal
Chatroom / IM	acked messaging, per-client IDs, auto-reconnect	examples/chatroom
Message queue	topics, ack, async publish	examples/mq
TCP relay / proxy	net.Conn-compatible streams over a control plane	examples/relay
API gateway / sidecar	bidirectional RPC + multiplexing + client identity	build directly on End

Architecture

Three layers — Connection (physical TCP, heartbeat, FSM), Multiplexer / Dialogue (logical streams, routing, write scheduling), and Application (RPC and messaging semantics) — let Geminio ship one unified End while keeping each concern isolated and testable. Deep dive in docs/MULTIPLEXING.md.

Benchmarks

Intel Core i5-6267U @ 2.90 GHz (2016 dual-core laptop):

BenchmarkMessage-4     10117   112584 ns/op   1164 MB/s
BenchmarkEnd-4         11644    98586 ns/op   1329 MB/s
BenchmarkStream-4      12301    96955 ns/op   1351 MB/s
BenchmarkRPC-4          6960   165384 ns/op    792 MB/s

~1.3 GB/s on streams, ~790 MB/s on end-to-end RPC round-trips — on a ten-year-old laptop CPU. Run make bench on your own box.

Documentation

• Usage guide — docs/USAGE.md
• API reference — pkg.go.dev/github.com/singchia/geminio
• Runnable examples — examples/
• Design deep dive — docs/MULTIPLEXING.md
• Roadmap — ROADMAP.md

Contributing

PRs and issues are welcome. See CONTRIBUTING.md. In short: one feature per PR, tests alongside code, run make test before submitting.

License

Apache 2.0 — © Austin Zhai, 2023–2030.

---

Made with OSS Insight

Documentation

Index

• type Call
• type End
• type HijackRPC
• type Message
• type Messager
• type MethodRPC
• type Multiplexer
• type Publish
• type RPC
• type RPCer
• type Raw
• type RawRPCMessager
• type Request
• type Response
• type Side
• type Stream

Types

type Call

type Call struct {
	Method   string
	Request  Request
	Response Response
	Error    error
	Done     chan *Call
}

for async RPC

type End

type End interface {
	// End is a default stream with streamID 1
	// Close on default stream will close all from the End
	Stream
	// End is a stream multiplexer
	Multiplexer

	// End is a net.Listener
	// Accept is a wrapper for AcceptStream
	// Addr is a wrapper for LocalAddr
	net.Listener

	Close() error
}

type HijackRPC

type HijackRPC func(context.Context, string, Request, Response)

hijack rpc functions

type Message

type Message interface {
	// to tell peer received or errored
	Done() error
	Error(err error) error

	// those meta info shouldn't be changed
	ID() uint64
	StreamID() uint64
	ClientID() uint64
	Timeout() time.Duration
	Topic() string // empty if not set
	// consistency protocol
	Cnss() options.Cnss
	// application data
	Data() []byte
	// custom data
	Custom() []byte

	// those Set operations must be accomplish before Publish
	SetTimeout(timeout time.Duration)
	SetCustom(data []byte)
	SetTopic(topic string)
	SetClientID(clientID uint64)
	SetStreamID(streamID uint64)
}

type Messager

type Messager interface {
	NewMessage(data []byte, opts ...*options.NewMessageOptions) Message

	Publish(ctx context.Context, msg Message, opts ...*options.PublishOptions) error
	PublishAsync(ctx context.Context, msg Message, ch chan *Publish, opts ...*options.PublishOptions) (*Publish, error)
	Receive(ctx context.Context) (Message, error)
}

type MethodRPC

type MethodRPC struct {
	Method string
	RPC    RPC
}

type Multiplexer

type Multiplexer interface {
	OpenStream(opts ...*options.OpenStreamOptions) (Stream, error)
	AcceptStream() (Stream, error)
	ListStreams() []Stream
}

Stream multiplexer

type Publish

type Publish struct {
	Message Message
	Error   error
	Done    chan *Publish
}

for async Publish

type RPC

type RPC func(context.Context, Request, Response)

rpc functions

type RPCer

type RPCer interface {
	NewRequest(data []byte, opts ...*options.NewRequestOptions) Request

	Call(ctx context.Context, method string, req Request, opts ...*options.CallOptions) (Response, error)
	CallAsync(ctx context.Context, method string, req Request, ch chan *Call, opts ...*options.CallOptions) (*Call, error)
	Register(ctx context.Context, method string, rpc RPC) error
	// Hijack rpc from remote
	Hijack(rpc HijackRPC, opts ...*options.HijackOptions) error
}

type Raw

type Raw net.Conn

type RawRPCMessager

type RawRPCMessager interface {
	// raw
	Raw
	// rpc
	RPCer
	// message
	Messager
}

type Request

type Request interface {
	// those meta info shouldn't be changed
	ID() uint64
	StreamID() uint64
	ClientID() uint64
	Method() string
	Timeout() time.Duration

	// application data
	Data() []byte
	// custom data
	Custom() []byte

	SetTimeout(timeout time.Duration)
	SetCustom([]byte)
	SetClientID(clientID uint64)
	SetStreamID(streamID uint64)
}

RPC releated

type Response

type Response interface {
	// those meta info shouldn't be changed
	ID() uint64
	StreamID() uint64
	ClientID() uint64
	Method() string

	// application data
	Data() []byte
	Error() error
	// custom data
	Custom() []byte

	SetData([]byte)
	SetError(error)
	SetCustom([]byte)
	SetClientID(clientID uint64)
	SetStreamID(streamID uint64)
}

type Side

type Side int

const (
	InitiatorSide Side = 0
	RecipientSide Side = 1
)

type Stream

type Stream interface {
	// a stream is a geminio
	RawRPCMessager
	// meta info for a stream
	StreamID() uint64
	ClientID() uint64
	Meta() []byte
	Side() Side
	Peer() string
}