lowlevel

package
v0.2.4 Latest Latest
Warning

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

 Go to latest Published: Nov 5, 2025 License: ISC Imports: 25 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrNoRtdtSessToRotateCookie = errors.New("no RTDT session to rotate cookie")

	ErrAlreadyPendingMantainRTDTSess = errors.New("already pending to maintain this session")
)

Functions

This section is empty.

Types

type AckError 

type AckError struct {
	ErrorStr  string
	ErrorCode int
	NonFatal  bool
}

AckError is an error generated when the server sends an Acknowledge message with an embedded Error message.

Is is also used in client code to signal a given pushed message was processed with an error.

func (AckError) Error 

func (err AckError) Error() string

func (AckError) Is 

func (err AckError) Is(target error) bool

func (AckError) ToAck 

func (err AckError) ToAck(ack *rpc.Acknowledge)

ToAck copies this error to the given Acknowledge msg.

type ConnKeeper 

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

ConnKeeper maintains an open connection to a server. Whenever the connection to the server closes, it attempts to re-connect. Only a single connection is kept online at any one time.

Fully kx'd server sessions are emitted via NextSession().

func NewConnKeeper 

func NewConnKeeper(cfg ConnKeeperCfg) *ConnKeeper

func (*ConnKeeper) AddKnownServerCerts added in v0.2.4 

func (ck *ConnKeeper) AddKnownServerCerts(tlsCert []byte, spid zkidentity.PublicIdentity)

AddKnownServerCerts adds a set of servers as already known. Whenever the ConnKeeper connects to an unknown server, it will ask for user confirmation.

func (*ConnKeeper) GoOnline 

func (ck *ConnKeeper) GoOnline()

GoOnline instructs the ConnKeeper to keep attempting connections to the server.

func (*ConnKeeper) NextSession 

func (ck *ConnKeeper) NextSession(ctx context.Context) clientintf.ServerSessionIntf

NextSession blocks until a session is available or the context is canceled. Note this returns nil in two situations: if the last session failed and is now offline or if the context is canceled.

func (*ConnKeeper) RemainOffline 

func (ck *ConnKeeper) RemainOffline()

RemainOffline asks the ConnKeeper to disconnect from the current session (if there is one) and to remain offline until GoOnline() is called.

func (*ConnKeeper) Run 

func (ck *ConnKeeper) Run(ctx context.Context) error

Run runs the services of this conn keeper.

type ConnKeeperCfg 

type ConnKeeperCfg struct {
	PC             clientintf.PaymentClient
	Dialer         clientintf.Dialer
	CertConf       clientintf.CertConfirmer
	PingInterval   time.Duration
	ReconnectDelay time.Duration
	Log            slog.Logger
	LogPings       bool

	// Passed to created serverSession instances (see there for reference).
	PushedRoutedMsgsHandler func(msg *rpc.PushRoutedMessage) error

	// OnUnwelcomeError is called when a connection attempt is rejected
	// due to a protocol negotiation error. This usually means the client
	// needs to be upgraded. This is called concurrently to the connection
	// attempts, therefore it should not block for long.
	OnUnwelcomeError func(err error)
}

type CreateRTDTSessionResult added in v0.2.4 

type CreateRTDTSessionResult struct {
	// SessionRV is the final session RV.
	SessionRV zkidentity.ShortID

	// SessionCookie is an opaque cookie to send to manage the session.
	SessionCookie []byte
}

CreateRTDTSessionResult is the result of a CreateSession call.

type ErrRVAlreadySubscribed 

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

func (ErrRVAlreadySubscribed) Error 

func (err ErrRVAlreadySubscribed) Error() string

func (ErrRVAlreadySubscribed) Is 

func (err ErrRVAlreadySubscribed) Is(target error) bool

type ErrRVAlreadyUnsubscribed 

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

func (ErrRVAlreadyUnsubscribed) Error 

func (err ErrRVAlreadyUnsubscribed) Error() string

func (ErrRVAlreadyUnsubscribed) Is 

func (err ErrRVAlreadyUnsubscribed) Is(target error) bool

type OutboundRM 

type OutboundRM interface {
	EncryptedLen() uint32
	EncryptedMsg() (RVID, []byte, error)
	Priority() uint
	PaidForRM(int64, int64)
}

OutboundRM is the interface for sending routed messages via the rmq.

type RMQ 

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

RMQ is a queue for sending RoutedMessages (RMs) to the server. The rmq supports a flickering server connection: any unsent RMs are queued (FIFO style) until a new server session is bound via `bindToSession`.

Sending an RM only fails when the rmq is shutting down or the rm failed to encrypt itself.

func NewRMQ 

func NewRMQ(log slog.Logger, db RMQDB) *RMQ

func (*RMQ) BindToSession 

func (q *RMQ) BindToSession(sess clientintf.ServerSessionIntf)

BindToSession binds the rmq to the specified server session. Queued and new messages will be sent via this server until it is removed or the rmq stops.

func (*RMQ) Len 

func (q *RMQ) Len() (int, int)

Len returns the current number of outstanding messages in the RMQs enqueue loop and send loop.

func (*RMQ) MaxMsgSize added in v0.1.10 

func (q *RMQ) MaxMsgSize() uint32

MaxMsgSize returns the current max message size of the RMQ.

func (*RMQ) QueueRM 

func (q *RMQ) QueueRM(orm OutboundRM, replyChan chan error) error

QueueRM enqueues the given RM to be sent to the server as soon as possible. Returns when the rm has been queued to be sent.

replyChan is written to when the RM has been received by server (which is determined when the RMQ receives the corresponding server ack) or if the rmq is stopping.

func (*RMQ) Run 

func (q *RMQ) Run(ctx context.Context) error

Run the services of this rmq. Must only be called once.

func (*RMQ) SendRM 

func (q *RMQ) SendRM(orm OutboundRM) error

SendRM sends the given routed message to the server whenever possible. It returns when the RM has been successfully written and acknowledged as received by the server.

func (*RMQ) TimingStats added in v0.1.2 

func (q *RMQ) TimingStats() []timestats.Quantile

TimingStats returns the latest timing stats for the RMQ.

type RMQDB added in v0.1.4 

type RMQDB interface {
	// StoreRVPaymentAttempt should store that an attempt to pay to push
	// to the given RV is being made with the given invoice.
	StoreRVPaymentAttempt(RVID, string, time.Time) error

	// RVHasPaymentAttempt should return the invoice and time that an
	// attempt to pay to push to the RV was made (i.e. it returns the
	// invoice and time saved on a prior call to StoreRVPaymentAttempt).
	RVHasPaymentAttempt(RVID) (string, time.Time, error)

	// DeleteRVPaymentAttempt removes the prior attempt to pay for the given
	// RV.
	DeleteRVPaymentAttempt(RVID) error
}

type RTDTManagerHandlers added in v0.2.4 

type RTDTManagerHandlers interface {
	JoinedLiveSession(rtSess *rtdtclient.Session, rv zkidentity.ShortID)
	RefreshedAllowance(rv zkidentity.ShortID, addAllowance uint64)
}

RTDTManagerHandlers are callbacks needed by the RTDT session manager.

type RTDTSessionManager added in v0.2.4 

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

RTDTSessionManager performs RTDT operations on a brserver. These include:

  • Creating sessions
  • Obtaining appointment cookies for administered sessions
  • Obtaining and maintaining join cookies for live sessions.

func NewRTDTSessionManager added in v0.2.4 

func NewRTDTSessionManager(rtc *rtdtclient.Client, handlers RTDTManagerHandlers, log slog.Logger) *RTDTSessionManager

NewRTDTSessionManager creates a new RTDT session manager.

func (*RTDTSessionManager) BindToSession added in v0.2.4 

func (rtsm *RTDTSessionManager) BindToSession(sess clientintf.ServerSessionIntf)

BindToSession sets the remote brserver connection.

func (*RTDTSessionManager) BytesWritten added in v0.2.4 

func (rtsm *RTDTSessionManager) BytesWritten(sess *rtdtclient.Session, n int)

BytesWritten should be used as a callback on the rtdtclient.Client instance to track data sent remotely to the server.

func (*RTDTSessionManager) CreateSession added in v0.2.4 

func (rtsm *RTDTSessionManager) CreateSession(size uint16) (*CreateRTDTSessionResult, error)

CreateSession attempts to create a RTDT session in brserver.

func (*RTDTSessionManager) ForceUnmaintainSession added in v0.2.4 

func (rtsm *RTDTSessionManager) ForceUnmaintainSession(sessRV *zkidentity.ShortID) error

ForceUnmaintainSession forcibly stops maintaining this session alive. This usually happens because the client was kicked from the session or had some other connection error.

func (*RTDTSessionManager) GetAppointCookies added in v0.2.4 

func (rtsm *RTDTSessionManager) GetAppointCookies(req *rpc.GetRTDTAppointCookies) (*rpc.GetRTDTAppointCookiesReply, error)

GetAppointCookies obtains appointment cookies for a session.

func (*RTDTSessionManager) LeaveSession added in v0.2.4 

func (rtsm *RTDTSessionManager) LeaveSession(sessRV *zkidentity.ShortID) error

LeaveSession stops maintaining the given session live. This does NOT make the client leave the actual RTDT sesssion, only stops obtaining new updated allowance for publishing.

func (*RTDTSessionManager) MaintainSession added in v0.2.4 

func (rtsm *RTDTSessionManager) MaintainSession(sessRV zkidentity.ShortID, sess *rpc.AppointRTDTServer,
	publisherKey *zkidentity.FixedSizeSymmetricKey, size uint32, peerID rpc.RTDTPeerID) error

MaintainSession joins and maintains a live session. This includes obtaining new join cookies with updated publishing allowance.

func (*RTDTSessionManager) RotateAppointCookie added in v0.2.4 

func (rtsm *RTDTSessionManager) RotateAppointCookie(sessRV *zkidentity.ShortID, newAppointCookie []byte) error

RotateAppointCookie replaces the appointment cookie for the given session.

func (*RTDTSessionManager) Run added in v0.2.4 

func (rtsm *RTDTSessionManager) Run(ctx context.Context) error

type RVBlob 

type RVBlob struct {
	Decoded  []byte
	ID       RVID
	ServerTS time.Time
}

RVBlob is a decoded blob received from the server at a specific RV point.

type RVHandler 

type RVHandler func(blob RVBlob) error

type RVID 

type RVID = ratchet.RVPoint

type RVManager 

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

RVManager keeps track of the various rendezvous points that should be registered on a remote server and what to do when RoutedMessages are received on the registered points.

Values should not be reused once their run() method returns.

func NewRVManager 

func NewRVManager(log slog.Logger, db RVManagerDB, subsDelayer func() <-chan time.Time, subDoneCB func()) *RVManager

func (*RVManager) BindToSession 

func (rmgr *RVManager) BindToSession(sess clientintf.ServerSessionIntf)

BindToSession binds the rendezvous manager to the specified server session.

Note: the rendezvous manager assumes the given session has been setup such that its `pushedRoutedMsgsHandler` calls the manager's `handlePushedRMs`.

func (*RVManager) FetchPrepaidRV added in v0.1.7 

func (rmgr *RVManager) FetchPrepaidRV(ctx context.Context, rdzv RVID) (RVBlob, error)

FetchPrepaidRV attempts to fetch the specified RV from the server without paying for it. For this to work with a server that expects payment, the RV must have been pre-paid already.

The provided ctx can be canceled to account for the fact that the RV may not actually exist in the server.

func (*RVManager) HandlePushedRMs 

func (rmgr *RVManager) HandlePushedRMs(prm *rpc.PushRoutedMessage) error

HandlePushedRMs is called via a bound session's `pushedRoutedMsgsHandler` whenever routed messages are pushed from server to client.

The received message is only ack'd after this function returns.

func (*RVManager) IsUpToDate 

func (rmgr *RVManager) IsUpToDate() bool

IsUpToDate returns true if the the manager has sent all updates to the remote server and the server has ack'd them.

func (*RVManager) PrepayRVSub added in v0.1.7 

func (rmgr *RVManager) PrepayRVSub(rdzv RVID, subPaid SubPaidHandler) error

PrepayRVSub pays for the specified RV in the server but does not subscribe to it.

func (*RVManager) Run 

func (rmgr *RVManager) Run(ctx context.Context) error

Run runs the rendezvous manager services. A given RVManager instance should not be reused once its run method returns.

func (*RVManager) Sub 

func (rmgr *RVManager) Sub(rdzv RVID, handler RVHandler, subPaid SubPaidHandler) error

Sub informs the manager to subscribe to the given rendezvous point and to call handler once a message is received in the given point.

Note that handler might never be called if the manager is stopped and it might be called multiple times if the rendezvous is registered and pushed multiple times.

The handler is called in the main goroutine for the RVManager, so it blocks further RV processing until it returns. Callers should arrange to spawn new goroutines if the handler will perform significant work.

func (*RVManager) Unsub 

func (rmgr *RVManager) Unsub(rdzv RVID) error

Unsub unsubscribes from the given rendezvous point.

type RVManagerDB 

type RVManagerDB interface {
	// UnpaidRVs filters the list of RVs, returning the ones that haven't
	// been paid yet.
	UnpaidRVs(rvs []RVID, expirationDays int) ([]RVID, error)

	// SavePaidRVs saves the specified list of RVs as paid.
	SavePaidRVs(rvs []RVID) error

	// MarkRVUnpaid marks the specified RV as unpaid in the DB.
	MarkRVUnpaid(rv RVID) error
}

RVManagerDB abstracts the necessary functions that the RV manager needs from the DB.

type SubPaidHandler 

type SubPaidHandler func(amount, fees int64)

SubPaidHandler is a callback type for tracking payment for subscribing to an RV.

type UnwelcomeError 

type UnwelcomeError struct {
	Reason string
}

UnwelcomeError is an error generated when the server responds with an Unwelcome message during the welcome stage of connection setup.

func (UnwelcomeError) Error 

func (err UnwelcomeError) Error() string

func (UnwelcomeError) Is 

func (err UnwelcomeError) Is(target error) bool

Source Files

View all Source files

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.