types

package
v30.0.0 Latest Latest
Warning

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

 Go to latest Published: Jul 16, 2026 License: Apache-2.0 Imports: 30 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DefaultSubscriptionBufferSize    = 32
	DefaultWsMaxConnections          = 200
	DefaultGrpcMaxConnections        = 200
	DefaultMaxSubscriptionsPerClient = 100

	// ModuleName defines the module name
	ModuleName = "stream"
)

Variables

View Source 
var (
	ErrCircuitBreakerOpen         = errorsmod.Register(ModuleName, 1, "circuit breaker open")
	ErrCircuitBreakerUnknownState = errorsmod.Register(ModuleName, 2, "unknown circuit breaker state")
	ErrConnectionLimit            = errorsmod.Register(ModuleName, 3, "connection limit reached")
	ErrSubscriptionLimit          = errorsmod.Register(ModuleName, 4, "subscription limit reached")
	ErrMissingModuleOrMethod      = errorsmod.Register(ModuleName, 10, "module and method are required")
	ErrStreamUnavailable          = errorsmod.Register(ModuleName, 11, "stream not available")
	ErrDescriptorRequired         = errorsmod.Register(ModuleName, 12, "descriptor is required")
	ErrUnknownParameter           = errorsmod.Register(ModuleName, 13, "unknown parameter")
	ErrMissingPathParameter       = errorsmod.Register(ModuleName, 14, "missing path parameter")
	ErrMissingQueryParameter      = errorsmod.Register(ModuleName, 15, "missing query parameter")
)
View Source 
var (
	ErrInvalidLengthQuery        = fmt.Errorf("proto: negative length found during unmarshaling")
	ErrIntOverflowQuery          = fmt.Errorf("proto: integer overflow")
	ErrUnexpectedEndOfGroupQuery = fmt.Errorf("proto: unexpected end of group")
)
View Source 
var Query_serviceDesc = _Query_serviceDesc

Functions

func RegisterQueryServer 

func RegisterQueryServer(s grpc1.Server, srv QueryServer)

func RunStream 

func RunStream(ctx context.Context, params RunStreamParams) error

RunStream orchestrates the full lifecycle of a streaming session. It performs connection accounting, delivers the initial snapshot and reacts to subsequent stream events until the context is cancelled.

func SanitizeLog 

func SanitizeLog(s string) string

SanitizeLog returns a copy of s with control characters (CR, LF, tab, other C0 bytes, DEL) replaced by '_'. Use on log fields whose value originates from a request header, subscription key, or other user-controlled source — passing such values to a structured logger unsanitised lets a caller forge log lines via embedded newlines (CWE-117 / log injection).

func SanitizeLogMap 

func SanitizeLogMap(m map[string]string) map[string]string

SanitizeLogMap returns a new map with both keys and values passed through SanitizeLog. Use it when logging a user-controlled string→string map.

Types

type ConnectionKind 

type ConnectionKind int

ConnectionKind identifies the transport type associated with a connection. 

const (
	ConnectionKindWebsocket ConnectionKind = iota
	ConnectionKindGRPC
)

type ConnectionMetadata 

type ConnectionMetadata struct {
	RemoteAddr   string
	ForwardedFor string
}

ConnectionMetadata captures identifying information for a subscriber connection.

type Dispatcher 

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

Dispatcher handles the routing of state events to subscriptions

func NewDispatcher 

func NewDispatcher(bufferSize int, registry *SubscriptionRegistry, logger log.Logger) *Dispatcher

NewDispatcher creates a new event dispatcher

func (*Dispatcher) Intake 

func (d *Dispatcher) Intake() chan<- encoding.StreamEvent

Intake returns the intake channel for producers to publish events.

func (*Dispatcher) IntakeCapacity 

func (d *Dispatcher) IntakeCapacity() int

IntakeCapacity returns the capacity of the intake channel buffer.

func (*Dispatcher) Start 

func (d *Dispatcher) Start(ctx context.Context)

Start begins the dispatcher event loop using the provided context.

func (*Dispatcher) Stop 

func (d *Dispatcher) Stop()

Stop stops the dispatcher

func (*Dispatcher) WaitForStop 

func (d *Dispatcher) WaitForStop()

WaitForStop waits for the dispatcher to fully stop

type QueryClient 

type QueryClient interface {
	// Stream is a dynamic streaming endpoint to add streaming support to any registered grpc unary query
	Stream(ctx context.Context, in *StreamDynamicRequest, opts ...grpc.CallOption) (Query_StreamClient, error)
}

QueryClient is the client API for Query service.

For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.

func NewQueryClient 

func NewQueryClient(cc grpc1.ClientConn) QueryClient

type QueryServer 

type QueryServer interface {
	// Stream is a dynamic streaming endpoint to add streaming support to any registered grpc unary query
	Stream(*StreamDynamicRequest, Query_StreamServer) error
}

QueryServer is the server API for Query service.

type Query_StreamClient 

type Query_StreamClient interface {
	Recv() (*StreamDynamicResponse, error)
	grpc.ClientStream
}

type Query_StreamServer 

type Query_StreamServer interface {
	Send(*StreamDynamicResponse) error
	grpc.ServerStream
}

type ResolvedStream 

type ResolvedStream struct {
	Descriptor *encoding.MethodDescriptor
	Fields     map[string]string
	Key        encoding.StreamEvent
}

ResolvedStream contains the metadata required to service a dynamic stream.

func ResolveStream 

func ResolveStream(
	registry *encoding.DynamicRegistry,
	module string,
	method string,
	params map[string]string,
) (*ResolvedStream, error)

ResolveStream locates the stream definition for the provided module/method pair and normalises the supplied parameters.

type RouterInvoker 

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

RouterInvoker routes gRPC queries through BaseApp's GRPCQueryRouter.

func NewRouterInvoker 

func NewRouterInvoker(app *baseapp.BaseApp, binaryCodec codec.BinaryCodec) (*RouterInvoker, error)

NewRouterInvoker creates a new RouterInvoker bound to the given BaseApp and codec.

func (*RouterInvoker) BuildRequest 

func (*RouterInvoker) BuildRequest(desc *encoding.MethodDescriptor, fields map[string]string) (protoiface.MessageV1, error)

BuildRequest constructs a protobuf request message using the provided method descriptor and field map.

func (*RouterInvoker) BuildResponse 

func (*RouterInvoker) BuildResponse(desc *encoding.MethodDescriptor) (protoiface.MessageV1, error)

BuildResponse allocates an empty response message for the provided method descriptor.

func (*RouterInvoker) Execute 

func (ri *RouterInvoker) Execute(desc *encoding.MethodDescriptor, fields map[string]string) (proto.Message, error)

Execute builds request/response messages and invokes the query using the latest committed height.

func (*RouterInvoker) Invoke 

func (ri *RouterInvoker) Invoke(req protoiface.MessageV1, resp protoiface.MessageV1) error

Invoke dispatches the request via the router using the latest committed height and fills the provided response message.

type RunStreamParams 

type RunStreamParams struct {
	// Registry manages subscriptions and connection accounting.
	Registry *SubscriptionRegistry
	// Key identifies the subscription to maintain.
	Key encoding.StreamEvent
	// ConnectionKind identifies the transport (gRPC or WebSocket).
	ConnectionKind ConnectionKind
	// ConnectionMeta carries optional metadata for connection tracking.
	ConnectionMeta ConnectionMetadata
	// Fetch is invoked to obtain the payload that should be sent to the client.
	// It is called once for the initial snapshot and again after each event.
	Fetch func(context.Context) (proto.Message, error)
	// Send delivers the payload to the downstream transport.
	Send func(proto.Message) error
	// OnEvent, when provided, is invoked for each matching StreamEvent.
	OnEvent func(encoding.StreamEvent)
	// Logger is optional and used for diagnostic messages.
	Logger log.Logger
	// OnConnect is invoked when the connection is successfully registered.
	OnConnect func(connectionID string) error
	// OnDisconnect is invoked after the connection is unregistered.
	OnDisconnect func(connectionID string)
}

RunStreamParams configures a streaming session that fans out updates from the subscription registry using the provided fetch and send hooks.

type StreamConfig 

type StreamConfig struct {
	SubscriptionBufferSize    uint32
	WsMaxConnections          uint32
	GrpcMaxConnections        uint32
	MaxSubscriptionsPerClient uint32
}

func LoadStreamConfig 

func LoadStreamConfig(homePath string) (StreamConfig, error)

type StreamDynamicRequest 

type StreamDynamicRequest struct {
	// module identifies the module that owns the target stream definition (e.g. bank, staking)
	Module string `protobuf:"bytes,1,opt,name=module,proto3" json:"module,omitempty"`
	// stream identifies the stream name within the module (e.g. balance, delegations)
	Method string `protobuf:"bytes,2,opt,name=method,proto3" json:"method,omitempty"`
	// Params contains the respective unary query parameters provided to the original module query
	Params map[string]string `` /* 153-byte string literal not displayed */
}

StreamDynamicRequest describes a dynamic stream subscription targeting a registered stream definition

func (*StreamDynamicRequest) Descriptor 

func (*StreamDynamicRequest) Descriptor() ([]byte, []int)

func (*StreamDynamicRequest) GetMethod 

func (m *StreamDynamicRequest) GetMethod() string

func (*StreamDynamicRequest) GetModule 

func (m *StreamDynamicRequest) GetModule() string

func (*StreamDynamicRequest) GetParams 

func (m *StreamDynamicRequest) GetParams() map[string]string

func (*StreamDynamicRequest) Marshal 

func (m *StreamDynamicRequest) Marshal() (dAtA []byte, err error)

func (*StreamDynamicRequest) MarshalTo 

func (m *StreamDynamicRequest) MarshalTo(dAtA []byte) (int, error)

func (*StreamDynamicRequest) MarshalToSizedBuffer 

func (m *StreamDynamicRequest) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*StreamDynamicRequest) ProtoMessage 

func (*StreamDynamicRequest) ProtoMessage()

func (*StreamDynamicRequest) Reset 

func (m *StreamDynamicRequest) Reset()

func (*StreamDynamicRequest) Size 

func (m *StreamDynamicRequest) Size() (n int)

func (*StreamDynamicRequest) String 

func (m *StreamDynamicRequest) String() string

func (*StreamDynamicRequest) Unmarshal 

func (m *StreamDynamicRequest) Unmarshal(dAtA []byte) error

func (*StreamDynamicRequest) XXX_DiscardUnknown 

func (m *StreamDynamicRequest) XXX_DiscardUnknown()

func (*StreamDynamicRequest) XXX_Marshal 

func (m *StreamDynamicRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*StreamDynamicRequest) XXX_Merge 

func (m *StreamDynamicRequest) XXX_Merge(src proto.Message)

func (*StreamDynamicRequest) XXX_Size 

func (m *StreamDynamicRequest) XXX_Size() int

func (*StreamDynamicRequest) XXX_Unmarshal 

func (m *StreamDynamicRequest) XXX_Unmarshal(b []byte) error

type StreamDynamicResponse 

type StreamDynamicResponse struct {
	// result carries the latest query response as an opaque protobuf Any.
	Result *any.Any `protobuf:"bytes,1,opt,name=result,proto3" json:"result,omitempty"`
}

StreamDynamicResponse returns the streamed data in its native protobuf representation.

func (*StreamDynamicResponse) Descriptor 

func (*StreamDynamicResponse) Descriptor() ([]byte, []int)

func (*StreamDynamicResponse) GetResult 

func (m *StreamDynamicResponse) GetResult() *any.Any

func (*StreamDynamicResponse) Marshal 

func (m *StreamDynamicResponse) Marshal() (dAtA []byte, err error)

func (*StreamDynamicResponse) MarshalTo 

func (m *StreamDynamicResponse) MarshalTo(dAtA []byte) (int, error)

func (*StreamDynamicResponse) MarshalToSizedBuffer 

func (m *StreamDynamicResponse) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*StreamDynamicResponse) ProtoMessage 

func (*StreamDynamicResponse) ProtoMessage()

func (*StreamDynamicResponse) Reset 

func (m *StreamDynamicResponse) Reset()

func (*StreamDynamicResponse) Size 

func (m *StreamDynamicResponse) Size() (n int)

func (*StreamDynamicResponse) String 

func (m *StreamDynamicResponse) String() string

func (*StreamDynamicResponse) Unmarshal 

func (m *StreamDynamicResponse) Unmarshal(dAtA []byte) error

func (*StreamDynamicResponse) XXX_DiscardUnknown 

func (m *StreamDynamicResponse) XXX_DiscardUnknown()

func (*StreamDynamicResponse) XXX_Marshal 

func (m *StreamDynamicResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*StreamDynamicResponse) XXX_Merge 

func (m *StreamDynamicResponse) XXX_Merge(src proto.Message)

func (*StreamDynamicResponse) XXX_Size 

func (m *StreamDynamicResponse) XXX_Size() int

func (*StreamDynamicResponse) XXX_Unmarshal 

func (m *StreamDynamicResponse) XXX_Unmarshal(b []byte) error

type StreamingListener 

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

StreamingListener implements the ABCIListener interface for the stream module

func NewStreamingListener 

func NewStreamingListener(
	intake chan<- encoding.StreamEvent,
	logger log.Logger,

) *StreamingListener

NewStreamingListener creates a new streaming listener

func (*StreamingListener) ListenCommit 

func (l *StreamingListener) ListenCommit(_ context.Context, _ abci.ResponseCommit, _ []*storetypes.StoreKVPair) error

ListenCommit implements the ABCIListener interface

func (*StreamingListener) ListenFinalizeBlock 

func (*StreamingListener) ListenFinalizeBlock(_ context.Context, _ abci.RequestFinalizeBlock, _ abci.ResponseFinalizeBlock) error

ListenFinalizeBlock implements the ABCIListener interface

type Subscriber 

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

Subscriber represents an active subscription

type SubscriptionRegistry 

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

SubscriptionRegistry manages active subscriptions

func NewSubscriptionRegistry 

func NewSubscriptionRegistry(logger log.Logger, cfg StreamConfig) *SubscriptionRegistry

NewSubscriptionRegistry creates a new subscription registry with the provided configuration.

func (*SubscriptionRegistry) AddConnectionSubscription 

func (r *SubscriptionRegistry) AddConnectionSubscription(connectionID string) bool

AddConnectionSubscription increments the subscription count for a connection if under limit.

func (*SubscriptionRegistry) Broadcast 

func (r *SubscriptionRegistry) Broadcast(data any)

Broadcast sends data to all active subscribers, regardless of their key. Temporary solution until event decoding is better supported

func (*SubscriptionRegistry) CanAcceptConnection 

func (r *SubscriptionRegistry) CanAcceptConnection(kind ConnectionKind) bool

CanAcceptConnection reports whether a new connection of the provided kind can be accepted.

func (*SubscriptionRegistry) CanAddConnectionSubscription 

func (r *SubscriptionRegistry) CanAddConnectionSubscription(connectionID string) bool

CanAddConnectionSubscription checks whether adding another subscription would exceed the per-connection limit.

func (*SubscriptionRegistry) CloseAll 

func (r *SubscriptionRegistry) CloseAll()

CloseAll closes all active subscriptions

func (*SubscriptionRegistry) ConnectionStats 

func (r *SubscriptionRegistry) ConnectionStats() (uint32, map[string]uint32)

ConnectionStats returns the total number of connections and per-connection subscription counts.

func (*SubscriptionRegistry) CurrentConnectionLimits 

func (r *SubscriptionRegistry) CurrentConnectionLimits() (wsMax, grpcMax, maxSubscriptionsPerConnection uint32)

CurrentConnectionLimits returns the configured limits and whether they are unbounded.

func (*SubscriptionRegistry) FanOut 

func (r *SubscriptionRegistry) FanOut(event encoding.StreamEvent, data any)

FanOut distributes an event to all subscribers with a matching key.

func (*SubscriptionRegistry) GetActiveConnections 

func (r *SubscriptionRegistry) GetActiveConnections() map[string]bool

GetActiveConnections returns a map of active connection IDs.

func (*SubscriptionRegistry) RegisterConnection 

func (r *SubscriptionRegistry) RegisterConnection(kind ConnectionKind, meta ConnectionMetadata) (string, error)

RegisterConnection registers a new connection of the provided kind and returns its UUID.

func (*SubscriptionRegistry) RemoveConnectionSubscription 

func (r *SubscriptionRegistry) RemoveConnectionSubscription(connectionID string)

RemoveConnectionSubscription decrements the subscription count for a connection.

func (*SubscriptionRegistry) Subscribe 

func (r *SubscriptionRegistry) Subscribe(ctx context.Context, key encoding.StreamEvent) (chan any, *Subscriber)

Subscribe adds a new subscription

func (*SubscriptionRegistry) UnregisterConnection 

func (r *SubscriptionRegistry) UnregisterConnection(connectionID string)

UnregisterConnection removes an active connection and updates metrics.

func (*SubscriptionRegistry) Unsubscribe 

func (r *SubscriptionRegistry) Unsubscribe(subscriber *Subscriber)

Unsubscribe removes a subscription

type UnimplementedQueryServer 

type UnimplementedQueryServer struct {
}

UnimplementedQueryServer can be embedded to have forward compatible implementations.

func (*UnimplementedQueryServer) Stream 

func (*UnimplementedQueryServer) Stream(req *StreamDynamicRequest, srv Query_StreamServer) error

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.