subscribe

Documentation

Index

• Constants
• Variables
• func FromAPI(f *api.BloomFilter) (*bloom.BloomFilter, error)
• func NewAuthorSubscription(authorPubs [][]byte, fp float64, rng *rand.Rand) (*api.Subscription, error)
• func NewFPSubscription(fp float64, rng *rand.Rand) (*api.Subscription, error)
• func NewReaderSubscription(readerPubs [][]byte, fp float64, rng *rand.Rand) (*api.Subscription, error)
• func NewSubscription(authorPubs [][]byte, authorFp float64, readerPubs [][]byte, readerFp float64, ...) (*api.Subscription, error)
• func ToAPI(f gob.GobEncoder) (*api.BloomFilter, error)
• type From
  • func NewFrom(params *FromParameters, logger *zap.Logger, out chan *KeyedPub) From
• type FromParameters
  • func NewDefaultFromParameters() *FromParameters
• type KeyedPub
• type PubReceipt
• type PublicationReceipts
• type RecentPublications
  • func NewRecentPublications(size uint32) (RecentPublications, error)
• type To
  • func NewTo(params *ToParameters, logger *zap.Logger, clientID ecid.ID, orgID ecid.ID, ...) To
• type ToParameters
  • func NewDefaultToParameters() *ToParameters

Constants

const (
	// DefaultNMaxSubscriptions is the default maximum number of subscriptions from clients
	// to support.
	DefaultNMaxSubscriptions = 64

	// DefaultEndSubscriptionProb is the default Bernoulli probability of ending a particular
	// subscription.
	DefaultEndSubscriptionProb = 1.0 / (1 << 20)
)

const (

	// DefaultNSubscriptionsTo is the default number of subscriptions to other peers to
	// maintain.
	DefaultNSubscriptionsTo = 10

	// DefaultFPRate is the default false positive rate for each subscription to another peer.
	DefaultFPRate = 1.0

	// DefaultTimeout is the default timeout for Subscribe requests. This is longer than a few
	// seconds because Subscribe responses are dependant on active publications, which may not
	// always be happening.
	DefaultTimeout = 30 * time.Minute

	// DefaultMaxErrRate is the default maximum allowed error rate for Subscribe requests and
	// received publications before a fatal error is thrown.
	DefaultMaxErrRate = 0.1

	// DefaultRecentCacheSize is the default recent publications LRU cache size.
	DefaultRecentCacheSize = 1 << 12
)

Variables

var (
	// ErrOutOfBoundsFPRate indicates when the false positive rate is not in (0, 1].
	ErrOutOfBoundsFPRate = errors.New("false positive rate out of (0, 1] bounds")

	// ErrNilPublicKeys indicates when either the author or reader public keys are nil.
	ErrNilPublicKeys = errors.New("nil public keys")
)

var ErrNotAcceptingNewSubscriptions = errors.New("not accepting new subscriptions")

ErrNotAcceptingNewSubscriptions indicates when new subscriptions are not being accepted.

Functions

func FromAPI

func FromAPI(f *api.BloomFilter) (*bloom.BloomFilter, error)

FromAPI converts an *api.BloomFilter to a *bloom.BloomFilter.

func NewAuthorSubscription

func NewAuthorSubscription(authorPubs [][]byte, fp float64, rng *rand.Rand) (
	*api.Subscription, error)

NewAuthorSubscription creates an *api.Subscription using the given author public keys and false positive rate with a 1.0 false positive rate for reader keys. This is useful when one doesn't want to filter on any reader keys.

func NewFPSubscription

func NewFPSubscription(fp float64, rng *rand.Rand) (*api.Subscription, error)

NewFPSubscription creates an *api.Subscription with the given false positive rate on the author keys and a 1.0 false positive rate on the reader keys.

func NewReaderSubscription

func NewReaderSubscription(readerPubs [][]byte, fp float64, rng *rand.Rand) (
	*api.Subscription, error)

NewReaderSubscription creates an *api.Subscription using the given reader public keys and false positive rate with a 1.0 false positive rate for author keys. This is useful when one doesn't want to filter on any author keys.

func NewSubscription

func NewSubscription(
	authorPubs [][]byte,
	authorFp float64,
	readerPubs [][]byte,
	readerFp float64,
	rng *rand.Rand,
) (*api.Subscription, error)

NewSubscription returns a new subscription with filters for the given author and reader public keys and false positive rates. Users should recall that the overall subscription false positive rate will be the product of the author and reader false positive rates.

func ToAPI

func ToAPI(f gob.GobEncoder) (*api.BloomFilter, error)

ToAPI converts a *bloom.BloomFilter (via narrower gob.GobEncoder) to an *api.BloomFilter.

Types

type From

type From interface {
	// Fanout copies messages from the outbound publication channel to the individual channels
	// of each subscriber.
	Fanout()

	// New creates a new subscriber channel, adds it to the fan-out, and returns it. If
	New() (chan *KeyedPub, chan struct{}, error)
}

From manages the fan-out from a main outbound channel to those of all the subscribers.

func NewFrom

func NewFrom(params *FromParameters, logger *zap.Logger, out chan *KeyedPub) From

NewFrom creates a new From instance that fans out from the given output channel.

type FromParameters

type FromParameters struct {
	// NMaxSubscriptions is the maximum number of concurrent subscriptions from other peers.
	NMaxSubscriptions uint32

	// EndSubscriptionProb is the Bernoulli probability of ending a particular subscription.
	EndSubscriptionProb float64
}

FromParameters define how the collection of subscriptions from other peer will be managed.

func NewDefaultFromParameters

func NewDefaultFromParameters() *FromParameters

NewDefaultFromParameters returns a *FromParameters object with default values.

type KeyedPub

type KeyedPub struct {
	// Key is the SHA-256 has of the value's bytes.
	Key id.ID

	// Value is the publication values.
	Value *api.Publication
}

KeyedPub couples a publication value and its key.

type PubReceipt

type PubReceipt struct {
	FromPub []byte
	Time    time.Time
}

PubReceipt represents a publication receipt from a peer (public key) at a particular time.

type PublicationReceipts

type PublicationReceipts struct {
	Value    *api.Publication
	Receipts []*PubReceipt
	// contains filtered or unexported fields
}

PublicationReceipts is a list of *PubReceipts for a given publication.

type RecentPublications

type RecentPublications interface {

	// Get returns the *PublicationsReceipts object and an indicator of whether the object
	// exists in the cache. The *PublicationsReceipts object is nil if it doesn't exist.
	Get(envelopeKey id.ID) (*PublicationReceipts, bool)

	// Add tracks that a given publication was received from the given peer public key and
	// returns whether that value has recently been seen before.
	Add(pvr *pubValueReceipt) bool

	// Len gives the number of items in the cache.
	Len() int
}

RecentPublications tracks publications recently received from peers with an internal LRU cache.

func NewRecentPublications

func NewRecentPublications(size uint32) (RecentPublications, error)

NewRecentPublications creates a RecentPublications LRU cache with a given size.

type To

type To interface {
	// Begin starts and runs the subscriptions to the peers. It runs indefinitely until either
	// a fatal error is encountered, or the subscriptions are gracefully stopped via End().
	Begin() error

	// End gracefully stops the active subscriptions and closes the new channel
	End()

	// Send sends a publication to the channel of received publications.
	Send(pub *api.Publication) error
}

To maintains active subscriptions to a collection of peers, merging their publications into a single, deduplicated stream.

func NewTo

func NewTo(
	params *ToParameters,
	logger *zap.Logger,
	clientID ecid.ID,
	orgID ecid.ID,
	csb client.SetBalancer,
	peerSigner client.Signer,
	orgSigner client.Signer,
	recent RecentPublications,
	new chan *KeyedPub,
) To

NewTo creates a new To instance, writing merged, deduplicated publications to the given new channel.

type ToParameters

type ToParameters struct {
	// NSubscriptions is the number of concurrent subscriptions maintained to other peers.
	NSubscriptions uint32

	// FPRate is the estimated false positive rate of the subscriptions.
	FPRate float32

	// Timeout is the timeout for each Subscribe request.
	Timeout time.Duration

	// MaxErrRate is the maximum allowed error rate for Subscribe requests and received
	// publications before a fatal error is thrown. This value is a running rate over a constant
	// history of responses (c.f., errQueueSize).
	MaxErrRate float32

	// RecentCacheSize is the size of the LRU cache used in deduplicating and grouping
	// publications.
	RecentCacheSize uint32
}

ToParameters define how the collection of subscriptions to other peers will be managed.

func NewDefaultToParameters

func NewDefaultToParameters() *ToParameters

NewDefaultToParameters returns a *ToParameters object with default values.