Documentation
¶
Overview ¶
Package outbox persists outgoing stream requests that failed because the destination node was offline and replays them, oldest-first, once that node is seen online again (via the pubsub discovery signal) or on startup.
Index ¶
- type Repo
- type RetryService
- func (s *RetryService) Close()
- func (s *RetryService) Enqueue(destNodeId, route string, payload []byte) error
- func (s *RetryService) FlushAllOnStart()
- func (s *RetryService) NotifyOnline(nodeId string)
- func (s *RetryService) NotifyOnlinePeer(info warpnet.WarpAddrInfo)
- func (s *RetryService) SetSender(sender Sender)
- type Sender
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type RetryService ¶
type RetryService struct {
// contains filtered or unexported fields
}
func NewRetryService ¶
func NewRetryService(ctx context.Context, repo Repo) *RetryService
func (*RetryService) Close ¶
func (s *RetryService) Close()
func (*RetryService) Enqueue ¶
func (s *RetryService) Enqueue(destNodeId, route string, payload []byte) error
Enqueue stores an undelivered request and marks its destination pending so a later NotifyOnline flushes it.
func (*RetryService) FlushAllOnStart ¶
func (s *RetryService) FlushAllOnStart()
FlushAllOnStart schedules a one-shot replay of every node that still has queued entries, covering the case where a peer is already online at startup and won't re-announce for a while.
func (*RetryService) NotifyOnline ¶
func (s *RetryService) NotifyOnline(nodeId string)
NotifyOnline is the pubsub-discovery hook: it is called for every peer seen on the discovery topic. It is O(1) and does no I/O when the node has nothing queued, so it is cheap to call on every gossip announcement.
func (*RetryService) NotifyOnlinePeer ¶
func (s *RetryService) NotifyOnlinePeer(info warpnet.WarpAddrInfo)
NotifyOnlinePeer matches the discovery.DiscoveryHandler signature so the retry service can be registered as an independent handler on the pubsub discovery topic (alongside, and without coupling to, the discovery service).
func (*RetryService) SetSender ¶
func (s *RetryService) SetSender(sender Sender)
SetSender wires the redelivery transport. Until it is set, flushes are no-ops, so it is safe to construct the service before the node is up.
type Sender ¶
type Sender interface {
Stream(nodeId warpnet.WarpPeerID, path stream.WarpRoute, data any) ([]byte, error)
}
Sender is the raw node stream used to replay queued requests. It is the node's own Stream (not the enqueuing GenericStream wrapper), so a failed redelivery does not re-enqueue and loop; the retry service owns the queue.
Source Files