Documentation
¶
Overview ¶
Package pubsub is a generic GossipSub publish/subscribe toolset over any cborx-envelope payload: a Publisher[T]/Follower[T] works for any T whose cbor-gen codec lives on *T (e.g. *core.FinalizedWithdrawal).
It is host-taking: the caller builds and owns the libp2p host and its connectivity; this package owns only the GossipSub instance, topic, and subscription.
Usage (constraint type inference fills the pointer type, so callers name only the value type):
pub, _ := pubsub.NewPublisher[core.FinalizedWithdrawal](ctx, h, topic, nil)
_ = pub.Publish(ctx, &core.FinalizedWithdrawal{...})
fol, _ := pubsub.NewFollower[core.FinalizedWithdrawal](ctx, h, topic, nil)
fol.SetHandler(func(fw *core.FinalizedWithdrawal) { ... })
go fol.Run(ctx)
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Follower ¶
type Follower[T any, M message[T]] struct { // contains filtered or unexported fields }
Follower subscribes to a topic on a caller-owned host and forwards each decoded value of type T to a Handler.
func NewFollower ¶
func NewFollower[T any, M message[T]](ctx context.Context, h host.Host, topic string, logger log.Logger) (*Follower[T, M], error)
NewFollower joins and subscribes to topic on h. A size-validator is attached before joining, so oversized messages are rejected by GossipSub and never reach the consume loop. Call Run to start consuming; register a handler with SetHandler before (or shortly after) Run. The caller owns h and its connectivity.
func (*Follower[T, M]) Close ¶
Close cancels the subscription and leaves the topic. It does not close the host — the caller owns that.
func (*Follower[T, M]) Run ¶
Run consumes the subscription until ctx is cancelled or the subscription closes. It blocks; run it in a goroutine.
func (*Follower[T, M]) SetHandler ¶
SetHandler installs the handler invoked for each decoded value. Safe to call concurrently with Run.
type Handler ¶
type Handler[T any] func(v *T)
Handler receives each decoded payload. The Follower calls it synchronously on the consume goroutine; a slow handler backs up incoming messages.
type Metrics ¶
type Metrics struct {
Delivered uint64 // payloads handed to the handler
DecodeErrors uint64 // envelope/CBOR decode failures
OversizeDrops uint64 // dropped for exceeding the size cap
// contains filtered or unexported fields
}
Metrics captures Follower counters for ops dashboards.
type Publisher ¶
type Publisher[T any, M message[T]] struct { // contains filtered or unexported fields }
Publisher joins a GossipSub topic and publishes values of type T. It does not subscribe: GossipSub only propagates once at least one subscriber is in the mesh, so a publisher-only node relies on its peers subscribing.
func NewPublisher ¶
func NewPublisher[T any, M message[T]](ctx context.Context, h host.Host, topic string, logger log.Logger) (*Publisher[T, M], error)
NewPublisher joins topic on h. The caller owns h and must keep it alive for the Publisher's lifetime.
func (*Publisher[T, M]) Close ¶
Close leaves the topic. It does not close the host — the caller owns that.
func (*Publisher[T, M]) Publish ¶
Publish emits v on the topic using the cborx V1 envelope. v is *T.
Source Files