etp

package module
v0.0.7 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Sep 1, 2026 License: MIT Imports: 11 Imported by: 0

README

go-etp

Go implementation of the Elum Transport Protocol and its application server layer.

The root package exposes App, compiled routing, middleware groups, peers, unified request/response and body handling. The wire/session implementation lives in internal/etp and is exposed through the root package where low-level integration is required.

The protocol includes:

  • binary frame header encode/decode;
  • payload encode/decode;
  • capabilities and schema constants;
  • session runtime;
  • auth frames;
  • heartbeat;
  • request/response frames;
  • chunked transfers;
  • ack/nack/retry;
  • flow control;
  • cancellation;
  • checksum;
  • graceful close;
  • transfer resume;
  • terminal transfer commit confirmation;
  • bounded send/receive queues and configurable per-session handler workers;
  • strict decoder, capability, rate-limit, and slowloris enforcement;
  • panic isolation at application callback boundaries;
  • stream slowloris guard.

The wire specification lives in RFC.md.

Import

import etp "github.com/elum-utils/go-etp"
app := etp.New(etp.Config{})
app.OnAuth(func(ctx context.Context, peer *etp.Peer, request etp.AuthRequest) (etp.AuthResult, error) {
	accountID, result, err := authenticate(ctx, request)
	if err == nil && result.OK {
		err = peer.SetRateLimitID(accountID)
	}
	return result, err
})
app.Use("message.*", func(next etp.Handler) etp.Handler {
	return func(ctx *etp.Context) error {
		return next(ctx)
	}
})
app.On("message.send", func(ctx *etp.Context) error {
	_, err := ctx.Respond(etp.SendOptions{Event: "message.sent", Body: ctx.Body})
	return err
})

SetRateLimitID aggregates request, frame, and byte quotas across every active connection of the authenticated user. If it is not called, quotas remain connection-local. The ID is set once and inactive buckets are retained briefly, so reconnecting cannot immediately reset an exhausted quota. OnProtocolEvent receives structured EventRateLimited details (LimitKind, Limit, RetryAfterMillis, and Violations) for application-level marking or bans.

Context and BodyView are valid only while the handler is running. Use EventCopy or Bytes when application code keeps data after the callback. Passing ctx.Body to Send, Request, or Respond safely selects an in-memory or streaming transfer and retains spooled files until completion. Asynchronous Data and metadata are owned before the method returns. Explicit Reader values must remain valid until MessageHandle.Done closes.

Adapters

Adapters are independent nested modules, so an application downloads only the network stack it imports:

  • github.com/elum-utils/go-etp/adapters/gorilla
  • github.com/elum-utils/go-etp/adapters/fiber
  • github.com/elum-utils/go-etp/adapters/nbio
  • github.com/elum-utils/go-etp/adapters/tcp
  • github.com/elum-utils/go-etp/adapters/tls
  • github.com/elum-utils/go-etp/adapters/quic
  • github.com/elum-utils/go-etp/adapters/webtransport

Large bodies are selected automatically by Session.Send, Request, and Respond: inline payloads use one request frame while larger or streaming bodies use transfer begin/data/end frames.

Test

go test ./...
go test -race ./...

Each adapter is tested from its own module directory.

Documentation

Index

Constants

View Source
const (
	RoleClient          = protocol.RoleClient
	RoleServer          = protocol.RoleServer
	DefaultChunkSize    = protocol.DefaultChunkSize
	HeaderSize          = protocol.HeaderSize
	MaxFrameBytes       = protocol.MaxFrameBytes
	MaxPooledFrameBytes = protocol.MaxPooledFrameBytes
	WireVersion         = protocol.WireVersion

	FrameData     = protocol.FrameData
	FrameRequest  = protocol.FrameRequest
	FrameResponse = protocol.FrameResponse
	FrameHello    = protocol.FrameHello
	FrameHelloAck = protocol.FrameHelloAck

	FlagFirst   = protocol.FlagFirst
	FlagLast    = protocol.FlagLast
	FlagControl = protocol.FlagControl

	SchemaTextMessage = protocol.SchemaTextMessage
	SchemaEvent       = protocol.SchemaEvent
	SchemaHello       = protocol.SchemaHello

	PriorityCritical = protocol.PriorityCritical
	ChannelControl   = protocol.ChannelControl

	ContentFile  = protocol.ContentFile
	ContentMedia = protocol.ContentMedia

	DefaultCapabilities = protocol.DefaultCapabilities
	AllCapabilities     = protocol.AllCapabilities

	CapabilityTransfers       = protocol.CapabilityTransfers
	CapabilityCancel          = protocol.CapabilityCancel
	CapabilityAck             = protocol.CapabilityAck
	CapabilityNack            = protocol.CapabilityNack
	CapabilityHeartbeat       = protocol.CapabilityHeartbeat
	CapabilityTransferSHA256  = protocol.CapabilityTransferSHA256
	CapabilityFlowControl     = protocol.CapabilityFlowControl
	CapabilitySlowlorisGuard  = protocol.CapabilitySlowlorisGuard
	CapabilityProtocolEvents  = protocol.CapabilityProtocolEvents
	CapabilityRequestResponse = protocol.CapabilityRequestResponse
	CapabilityGracefulClose   = protocol.CapabilityGracefulClose
	CapabilityTransferResume  = protocol.CapabilityTransferResume
	CapabilityTransferCommit  = protocol.CapabilityTransferCommit
	CapabilityRateLimits      = protocol.CapabilityRateLimits

	ChecksumOff            = protocol.ChecksumOff
	ChecksumTransferSHA256 = protocol.ChecksumTransferSHA256

	SessionNew         = protocol.SessionNew
	SessionEstablished = protocol.SessionEstablished
	SessionClosed      = protocol.SessionClosed
	SessionFailed      = protocol.SessionFailed

	EventProtocolViolation = protocol.EventProtocolViolation
	EventSlowloris         = protocol.EventSlowloris
	EventTransferFailed    = protocol.EventTransferFailed
	EventTransferUnknown   = protocol.EventTransferUnknown
	EventTransferCanceled  = protocol.EventTransferCanceled
	EventWriteFailed       = protocol.EventWriteFailed
	EventNackReceived      = protocol.EventNackReceived
	EventAckTimeout        = protocol.EventAckTimeout
	EventAuthAccepted      = protocol.EventAuthAccepted
	EventAuthRejected      = protocol.EventAuthRejected
	EventAuthTimeout       = protocol.EventAuthTimeout
	EventAuthRequired      = protocol.EventAuthRequired
	EventErrorReceived     = protocol.EventErrorReceived
	EventGoAwayReceived    = protocol.EventGoAwayReceived
	EventCloseAckReceived  = protocol.EventCloseAckReceived
	EventHandlerFailed     = protocol.EventHandlerFailed
	EventRateLimited       = protocol.EventRateLimited
	EventChecksumMismatch  = protocol.EventChecksumMismatch

	RateLimitNone     = protocol.RateLimitNone
	RateLimitRequests = protocol.RateLimitRequests
	RateLimitFrames   = protocol.RateLimitFrames
	RateLimitBytes    = protocol.RateLimitBytes
)

Variables

View Source
var (
	ErrBodyTooLargeForBytes = errors.New("transport: body is not fully in memory")
	ErrBodyWriterClosed     = errors.New("transport: body writer is closed")
)
View Source
var (
	ErrRouterCompiled     = errors.New("transport: router is already compiled")
	ErrRouterNotCompiled  = errors.New("transport: router is not compiled")
	ErrRouteNotFound      = errors.New("transport: route not found")
	ErrRouteAlreadyExists = errors.New("transport: route already exists")
	ErrRoutePatternEmpty  = errors.New("transport: route pattern is empty")
	ErrGroupPrefixCount   = errors.New("transport: group accepts at most one prefix")
	ErrMiddlewareNil      = errors.New("transport: middleware is nil")
	ErrHandlerNil         = errors.New("transport: handler is nil")
)
View Source
var (
	ErrRateLimitIDEmpty   = errors.New("transport: rate limit id is empty")
	ErrRateLimitIDChanged = errors.New("transport: rate limit id cannot be changed")
)
View Source
var ErrNilFrameTransport = errors.New("transport: nil frame transport")

Functions

func EncodeEventMessageStringWithFieldsInto

func EncodeEventMessageStringWithFieldsInto(dst []byte, event string, data []byte, fields []TransferField) ([]byte, error)

func EncodeFrame

func EncodeFrame(frame Frame) ([]byte, error)

func EncodeFrameInto

func EncodeFrameInto(dst []byte, frame Frame) ([]byte, error)

func EncodeHelloMessage

func EncodeHelloMessage(hello Hello) []byte

func NewRouter

func NewRouter() *compiledRouter

Types

type Ack

type Ack = protocol.Ack

type Adapter

type Adapter interface {
	Name() string
	Serve(context.Context, *App) error
}

type App

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

func New

func New(config Config) *App

func (*App) Compile

func (a *App) Compile()

func (*App) Group

func (a *App) Group(prefix ...string) *Group

Group creates a route group. It accepts zero or one optional event prefix.

func (*App) MaxFrameBytes

func (a *App) MaxFrameBytes() uint32

func (*App) On

func (a *App) On(pattern string, handler Handler) error

func (*App) OnAuth

func (a *App) OnAuth(handler AuthHandler) error

OnAuth registers the server-side authentication handler and enables required ETP authentication. It must be registered before Compile.

func (*App) OnConnect

func (a *App) OnConnect(handler ConnectHandler) error

OnConnect registers a handler that runs after authentication and handshake. It must be registered before Compile.

func (*App) OnDisconnect

func (a *App) OnDisconnect(handler DisconnectHandler) error

OnDisconnect registers a handler that runs once when a peer disconnects. It must be registered before Compile.

func (*App) OnError

func (a *App) OnError(handler ErrorHandler) error

OnError registers a handler for errors returned by middleware, routes, and OnNotFound. Protocol-level events use OnProtocolEvent.

func (*App) OnNotFound

func (a *App) OnNotFound(handler Handler) error

OnNotFound registers a handler for incoming events without a route. It must be registered before Compile.

func (*App) OnProgress

func (a *App) OnProgress(handler ProgressHandler) error

OnProgress registers a handler for incoming and outgoing transfer progress.

func (*App) OnProtocolEvent

func (a *App) OnProtocolEvent(handler ProtocolEventHandler) error

OnProtocolEvent registers a handler for malformed frames, rate limits, authentication failures, transport events, and transfer state events.

func (*App) ServeTransport

func (a *App) ServeTransport(ctx context.Context, name string, transport protocol.FrameTransport) (*Peer, error)

func (*App) ServeTransportWithRemote

func (a *App) ServeTransportWithRemote(ctx context.Context, name string, remote string, transport protocol.FrameTransport) (*Peer, error)

func (*App) Use

func (a *App) Use(pattern string, middleware Middleware) error

type AuthAccept

type AuthAccept = protocol.AuthAccept

type AuthAttribute

type AuthAttribute = protocol.AuthAttribute

type AuthConfig

type AuthConfig = protocol.AuthConfig

type AuthHandler

AuthHandler validates a peer during the ETP authentication exchange. The peer exposes adapter and remote-address metadata needed by application authentication, such as an IP-bound session check.

Returning AuthResult{OK: false} rejects the connection without exposing the underlying validation error to the client.

type AuthReject

type AuthReject = protocol.AuthReject

type AuthRequest

type AuthRequest = protocol.AuthRequest

type AuthResult

type AuthResult = protocol.AuthResult

type Body

type Body interface {
	Size() int64
	Open() (io.ReadCloser, error)
	Bytes() ([]byte, error)
	View() ([]byte, bool)
	IsInline() bool
}

func NewBytesBody

func NewBytesBody(data []byte) Body

type Cancel

type Cancel = protocol.Cancel

type ChecksumMode

type ChecksumMode = protocol.ChecksumMode

type CloseConfig

type CloseConfig = protocol.CloseConfig

type CloseMessage

type CloseMessage = protocol.CloseMessage

type Config

type Config struct {
	Session            protocol.SessionConfig
	MaxMemoryBody      int64
	MaxPooledBodyBytes int
	OnError            ErrorHandler
	// RateLimitIdentityTTL retains an inactive authenticated user's bucket long
	// enough to prevent reconnects from resetting a recently exhausted quota.
	RateLimitIdentityTTL time.Duration
}

type ConnectHandler

type ConnectHandler func(context.Context, *Peer) error

ConnectHandler runs once after authentication and the ETP handshake succeed. Returning an error closes the connection.

type Context

type Context struct {
	context.Context
	App        *App
	Peer       *Peer
	Event      string
	RequestID  uint64
	TransferID uint64
	Fields     []Field
	Body       Body
	// contains filtered or unexported fields
}

Context is borrowed for one synchronous Handler call and must not be retained.

func (*Context) BodyView

func (c *Context) BodyView() ([]byte, bool)

BodyView returns body bytes without copying. The view is valid only until the handler returns.

func (*Context) Bytes

func (c *Context) Bytes() ([]byte, error)

func (*Context) EventCopy

func (c *Context) EventCopy() string

EventCopy returns an event name that remains valid after the handler returns.

func (*Context) Field

func (c *Context) Field(name string) string

func (*Context) Respond

func (c *Context) Respond(opts SendOptions) (MessageHandle, error)

type DeadlineStream

type DeadlineStream = protocol.DeadlineStream

type DisconnectHandler

type DisconnectHandler func(context.Context, *Peer, error)

DisconnectHandler runs once when an established peer disconnects. The cause is nil for a graceful close.

type ErrorHandler

type ErrorHandler func(*Context, error)

type ErrorMessage

type ErrorMessage = protocol.ErrorMessage

type EventMessage

type EventMessage = protocol.EventMessage

type EventMessageView

type EventMessageView = protocol.EventMessageView

func DecodeEventMessageView

func DecodeEventMessageView(payload []byte) (EventMessageView, error)

type Field

type Field = protocol.TransferField

type FlowControlConfig

type FlowControlConfig = protocol.FlowControlConfig

type Frame

type Frame = protocol.Frame

func DecodeFrameView

func DecodeFrameView(data []byte) (Frame, error)

func NewFrame

func NewFrame(frameType uint8, schemaID uint32, payload []byte) Frame

type FrameLease

type FrameLease = protocol.FrameLease

func InitFrameLease

func InitFrameLease(lease *FrameLease, data []byte, releaser FrameLeaseReleaser) *FrameLease

func NewFrameLease

func NewFrameLease(data []byte, release func([]byte)) *FrameLease

type FrameLeaseReleaser

type FrameLeaseReleaser = protocol.FrameLeaseReleaser

type FrameLimitTransport

type FrameLimitTransport = protocol.FrameLimitTransport

type FrameTransport

type FrameTransport = protocol.FrameTransport

type FrameTransportAdapter

type FrameTransportAdapter struct {
	AdapterName string
	Transport   protocol.FrameTransport
}

func (FrameTransportAdapter) Name

func (a FrameTransportAdapter) Name() string

func (FrameTransportAdapter) Serve

func (a FrameTransportAdapter) Serve(ctx context.Context, app *App) error

type GoAway

type GoAway = protocol.GoAway

type Group

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

func (*Group) Group

func (g *Group) Group(prefix ...string) *Group

Group creates a nested route group. It accepts zero or one optional event prefix.

func (*Group) On

func (g *Group) On(pattern string, handler Handler) error

func (*Group) Use

func (g *Group) Use(pattern string, middleware Middleware) error

type Handler

type Handler func(*Context) error

type HandlerConfig

type HandlerConfig = protocol.HandlerConfig
type Header = protocol.Header

func DecodeHeader

func DecodeHeader(data []byte) (Header, error)

type HeartbeatConfig

type HeartbeatConfig = protocol.HeartbeatConfig

type Hello

type Hello = protocol.Hello

type IncomingTransferAborter

type IncomingTransferAborter = protocol.IncomingTransferAborter

type IncomingTransferContextCloser

type IncomingTransferContextCloser = protocol.IncomingTransferContextCloser

type IncomingTransferContextWriter

type IncomingTransferContextWriter = protocol.IncomingTransferContextWriter

type IncomingTransferInfo

type IncomingTransferInfo = protocol.IncomingTransferInfo

type IncomingTransferSuspender

type IncomingTransferSuspender = protocol.IncomingTransferSuspender

type IncomingTransferWriter

type IncomingTransferWriter = protocol.IncomingTransferWriter

type LeasedFrameTransport

type LeasedFrameTransport = protocol.LeasedFrameTransport

type MessageHandle

type MessageHandle = protocol.MessageHandle

type MessageOptions

type MessageOptions = protocol.MessageOptions

type Middleware

type Middleware func(Handler) Handler

type MultiStreamTransport

type MultiStreamTransport = protocol.MultiStreamTransport

func NewMultiStreamTransport

func NewMultiStreamTransport(config MultiStreamTransportConfig) *MultiStreamTransport

type MultiStreamTransportConfig

type MultiStreamTransportConfig = protocol.MultiStreamTransportConfig

type Nack

type Nack = protocol.Nack

type NegotiatedFrameTransport

type NegotiatedFrameTransport = protocol.NegotiatedFrameTransport

type PayloadLimitConfig

type PayloadLimitConfig = protocol.PayloadLimitConfig

type Peer

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

func (*Peer) Adapter

func (p *Peer) Adapter() string

func (*Peer) Close

func (p *Peer) Close() error

func (*Peer) Context

func (p *Peer) Context() context.Context

Context is canceled when the peer session finishes.

func (*Peer) Identity

func (p *Peer) Identity() protocol.SessionIdentity

func (*Peer) RateLimitID

func (p *Peer) RateLimitID() string

RateLimitID returns the server-assigned identity used to aggregate limits. An empty value means that this peer is limited independently by connection.

func (*Peer) RemoteAddr

func (p *Peer) RemoteAddr() string

func (*Peer) Request

func (p *Peer) Request(ctx context.Context, opts SendOptions) (MessageHandle, error)

func (*Peer) Respond

func (p *Peer) Respond(ctx context.Context, requestID uint64, opts SendOptions) (MessageHandle, error)

func (*Peer) Send

func (p *Peer) Send(ctx context.Context, opts SendOptions) (MessageHandle, error)

func (*Peer) Session

func (p *Peer) Session() *protocol.Session

func (*Peer) SetRateLimitID

func (p *Peer) SetRateLimitID(id string) error

SetRateLimitID switches this peer from its connection-local bucket to the bucket shared by all connections with id. Call it only with an identity established by trusted server-side authentication. The identity is set-once so application code cannot accidentally evade limits by rotating keys.

type Progress

type Progress = protocol.Progress

type ProgressHandler

type ProgressHandler func(context.Context, *Peer, protocol.Progress)

ProgressHandler receives transfer progress for incoming and outgoing bodies. It is called synchronously by the session and should return quickly.

type ProtocolEvent

type ProtocolEvent = protocol.ProtocolEvent

type ProtocolEventHandler

type ProtocolEventHandler func(context.Context, *Peer, protocol.ProtocolEvent)

ProtocolEventHandler receives non-application protocol events. It is called synchronously by the session and should return quickly.

type RateLimitAdvertisement

type RateLimitAdvertisement = protocol.RateLimitAdvertisement

func DefaultRateLimitAdvertisement

func DefaultRateLimitAdvertisement() RateLimitAdvertisement

type RateLimitConfig

type RateLimitConfig = protocol.RateLimitConfig

type ReceiveConfig

type ReceiveConfig = protocol.ReceiveConfig

type RequestHandler

type RequestHandler = protocol.RequestHandler

type ResponseHandler

type ResponseHandler = protocol.ResponseHandler

type ResumeConfig

type ResumeConfig = protocol.ResumeConfig

type ResumeTransferOptions

type ResumeTransferOptions = protocol.ResumeTransferOptions

type Router

type Router interface {
	// On registers handler for one exact event name, for example
	// "control.workspace.get". Event names are combined with the optional
	// Group prefix when the router is compiled. It returns ErrRoutePatternEmpty
	// for an empty name, ErrRouteAlreadyExists for a duplicate route,
	// ErrHandlerNil for a nil handler, or ErrRouterCompiled after Compile.
	On(event string, handler Handler) error

	// Use registers middleware for matching routes. pattern may be an exact
	// event name, a namespace wildcard such as "control.*", or "*" for every
	// route in the current group. Middleware runs in registration order before
	// the matching handler. It returns ErrMiddlewareNil for a nil middleware or
	// ErrRouterCompiled after Compile.
	Use(pattern string, middleware Middleware) error

	// Group creates a nested route scope. It accepts zero arguments for a group
	// without a prefix, or one prefix such as "control". Routes registered on a
	// prefixed group are combined with that prefix: group.On("workspace.get", h)
	// registers "control.workspace.get". Middleware registered on parent groups
	// is inherited by child groups. Group panics with ErrRouterCompiled after
	// Compile or ErrGroupPrefixCount when more than one prefix is supplied.
	Group(prefix ...string) *Group
}

Router is the route-registration surface shared by App and Group. Controllers should depend on this interface rather than a concrete transport.

type SchedulerConfig

type SchedulerConfig = protocol.SchedulerConfig

type SendOptions

type SendOptions struct {
	Event string
	// Body selects an in-memory or streaming transfer automatically. It cannot
	// be combined with Data or Reader.
	Body Body
	// Data is copied if sending requires an asynchronous transfer.
	Data   []byte
	Fields []Field
	// Reader is consumed asynchronously and must remain valid until MessageHandle.Done closes.
	Reader      io.Reader
	Size        uint64
	Name        string
	Field       string
	Index       uint32
	ContentType uint32
	ChunkSize   int
	AckTimeout  time.Duration
	RetryLimit  int
}

type SendQueueConfig

type SendQueueConfig = protocol.SendQueueConfig

type Session

type Session = protocol.Session

func NewSession

func NewSession(t FrameTransport) *Session

func NewSessionWithConfig

func NewSessionWithConfig(t FrameTransport, config SessionConfig) *Session

type SessionConfig

type SessionConfig = protocol.SessionConfig

func DefaultClientConfig

func DefaultClientConfig() SessionConfig

func DefaultServerConfig

func DefaultServerConfig() SessionConfig

func DefaultSessionConfig

func DefaultSessionConfig(role string) SessionConfig

func NormalizeSessionConfig

func NormalizeSessionConfig(config SessionConfig) SessionConfig

type SessionIdentity

type SessionIdentity = protocol.SessionIdentity

type SessionState

type SessionState = protocol.SessionState

type SlowlorisConfig

type SlowlorisConfig = protocol.SlowlorisConfig

func DefaultSlowlorisConfig

func DefaultSlowlorisConfig() SlowlorisConfig

type StreamTransport

type StreamTransport = protocol.StreamTransport

func NewStreamTransport

func NewStreamTransport(conn net.Conn) *StreamTransport

func NewStreamTransportForStream

func NewStreamTransportForStream(stream DeadlineStream, guard SlowlorisConfig) *StreamTransport

func NewStreamTransportWithSlowlorisGuard

func NewStreamTransportWithSlowlorisGuard(conn net.Conn, guard SlowlorisConfig) *StreamTransport

type TextHandler

type TextHandler = protocol.TextHandler

type TransferBegin

type TransferBegin = protocol.TransferBegin

type TransferField

type TransferField = protocol.TransferField

type TransferHandle

type TransferHandle = protocol.TransferHandle

type TransferHandler

type TransferHandler = protocol.TransferHandler

type TransferOptions

type TransferOptions = protocol.TransferOptions

type TransferPart

type TransferPart = protocol.TransferPart

type TransferResume

type TransferResume = protocol.TransferResume

type TransferResumeDecision

type TransferResumeDecision = protocol.TransferResumeDecision

type TransferResumeStore

type TransferResumeStore = protocol.TransferResumeStore

type TransferResumeView

type TransferResumeView = protocol.TransferResumeView

type TransferState

type TransferState = protocol.TransferState

type TransferStateMessage

type TransferStateMessage = protocol.TransferStateMessage

type Window

type Window = protocol.Window

Directories

Path Synopsis
adapters
fiber module
internal
etp

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL