collect

type AsyncPubSub ¶

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

AsyncPubSub encapsulates pubsub, provides async methods to get subscribe messages. AsyncPubSub also manages the joined topics

func NewAsyncPubSub(h host.Host, opts ...TopicOpt) (apsub *AsyncPubSub, err error)

func (ap *AsyncPubSub) Close() error

func (ap *AsyncPubSub) LoadTopicItem(topic string, key interface{}) (value interface{}, err error)

func (ap *AsyncPubSub) Publish(ctx context.Context, topic string, data []byte) (err error)

func (ap *AsyncPubSub) SetTopicItem(topic string, key interface{}, value interface{}) error

func (ap *AsyncPubSub) Subscribe(topic string, handle TopicHandle) (err error)

func (ap *AsyncPubSub) Topics() (out []string)

func (ap *AsyncPubSub) Unsubscribe(topic string) (err error)

type BasicPubSubCollector ¶

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

BasicPubSubCollector implements of psc.BasicPubSubCollector Interface. It broadcasts request by libp2p.PubSub. When the remote nodes receive the request, they will try to dial the request node and return the response directly.

func NewBasicPubSubCollector(h host.Host, options ...InitOpt) (bpsc *BasicPubSubCollector, err error)

func (bpsc *BasicPubSubCollector) Close() (err error)

func (bpsc *BasicPubSubCollector) Join(topic string, opts ...JoinOpt) (err error)

func (bpsc *BasicPubSubCollector) Leave(topic string) (err error)

func (bpsc *BasicPubSubCollector) Publish(topic string, payload []byte, opts ...PubOpt) (err error)

type Conf ¶

type Conf struct {
	// Router is an option to select different router type
	// Router must be one of `basic`, `relay` and `intbfs`
	Router string `json:"router"`
	// ProtocolPrefix is the protocol name prefix
	ProtocolPrefix string `json:"protocol"`
	// RequestCacheSize .
	// RequestCache is used to store the request control message,
	// which is for response routing.
	RequestCacheSize int `json:"requestCacheSize"`
	// ResponseCacheSize .
	// ResponseCache is used to deduplicate the response.
	ResponseCacheSize int `json:"responseCacheSize"`
	// Fanout .
	Fanout int `json:"fanout"`
	// RandomFanout .
	RandomFanout int `json:"randomFanout"`
	// MaxHitsToSend
	MaxHitsToSend int `json:"maxHitsToSend"`
}

Conf is the static configuration of PubSubCollector

func MakeDefaultConf() Conf

type FinalRespHandler ¶

type FinalRespHandler func(context.Context, *Response)

FinalRespHandler is the callback function when the root node receiving a response. It will be called only in the root node. It will be called more than one time when the number of responses is larger than one.

type HostWires ¶ added in v0.0.9

type HostWires struct {
	host.Host
	// contains filtered or unexported fields
}

func NewHostWires(h host.Host) *HostWires

func (hw *HostWires) ClosedStream(network.Network, network.Stream)

func (hw *HostWires) Connected(n network.Network, c network.Conn)

func (hw *HostWires) Disconnected(n network.Network, c network.Conn)

func (hw *HostWires) ID() peer.ID

func (hw *HostWires) Listen(network.Network, ma.Multiaddr)

func (hw *HostWires) ListenClose(network.Network, ma.Multiaddr)

func (hw *HostWires) Neighbors() []peer.ID

func (hw *HostWires) OpenedStream(network.Network, network.Stream)

func (hw *HostWires) SendMsg(to peer.ID, data []byte) error

func (hw *HostWires) SetListener(wn WireListener)

func (hw *HostWires) SetMsgHandler(h MsgHandler)

type InitOpt ¶

type InitOpt func(*InitOpts) error

InitOpt is options used in NewBasicPubSubCollector

func WithConf(conf Conf) InitOpt

func WithLogger(l Logger) InitOpt

func WithRequestIDGenerator(ridFn ReqIDFn) InitOpt

type InitOpts ¶

type InitOpts struct {
	Conf    Conf
	ReqIDFn ReqIDFn
	MsgIDFn pubsub.MsgIDFn
	Logger  Logger
	Wires   TopicWires
}

InitOpts is options used in NewBasicPubSubCollector

func NewInitOpts(opts []InitOpt) (out *InitOpts, err error)

type IntBFS ¶ added in v0.0.9

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

IntBFS don't care about topic. IntBFS contains 5 parts: 1. query machanism 2. peer profiles 3. peer ranking 4. distance function 5. random perturbation

func NewIntBFS(wires Wires, opts *IntBFSOptions) (*IntBFS, error)

func (ib *IntBFS) Close() error

func (ib *IntBFS) HandlePeerDown(p peer.ID)

func (ib *IntBFS) HandlePeerUp(p peer.ID)

func (ib *IntBFS) Publish(data []byte, opts ...PubOpt) error

type IntBFSCollector ¶ added in v0.0.9

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

IntBFSCollector . We used a topic-defined overlay here.

func NewIntBFSCollector(h host.Host, opts ...InitOpt) (*IntBFSCollector, error)

func (ic *IntBFSCollector) Close() error

func (ic *IntBFSCollector) Join(topic string, opts ...JoinOpt) (err error)

func (ic *IntBFSCollector) Leave(topic string) (err error)

func (ic *IntBFSCollector) Publish(topic string, data []byte, opts ...PubOpt) error

type IntBFSOptions ¶ added in v0.0.9

type IntBFSOptions struct {
	Fanout        int
	RandomFanout  int
	MaxHitsToSend int
	ProfileFactory
	RequestHandler
	ReqIDFn
	Logger
	Topic string
}

IntBFSOptions .

func MakeDefaultIntBFSOptions() *IntBFSOptions

type Intermediate ¶

type Intermediate = pb.Intermediate

Intermediate type alias

type JoinOpt ¶

type JoinOpt func(*JoinOpts) error

JoinOpt is optional options in PubSubCollector.Join

func WithProfileFactory(pf ProfileFactory) JoinOpt

func WithRequestHandler(rqhandle RequestHandler) JoinOpt

type JoinOpts ¶

type JoinOpts struct {
	RequestHandler
	ProfileFactory
}

JoinOpts is the aggregated options

func NewJoinOptions(opts []JoinOpt) (out *JoinOpts, err error)

type Logger ¶

type Logger interface {
	Logf(level, format string, args ...interface{})
}

Logger . level is {"debug", "info", "warn", "error", "fatal"}; format and args are compatable with fmt.Printf.

func MakeDefaultLogger() Logger

type Message ¶

type Message = pubsub.Message

Message is type alias

type Msg ¶ added in v0.0.9

type Msg struct {
	*pb.Msg
	// contains filtered or unexported fields
}

type MsgHandler ¶ added in v0.0.9

type MsgHandler func(from peer.ID, data []byte)

MsgHandler .

type PeerSet ¶

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

func NewPeerSet() *PeerSet

func (ps *PeerSet) Add(p peer.ID)

func (ps *PeerSet) Del(p peer.ID)

func (ps *PeerSet) Equal(another *PeerSet) bool

func (ps *PeerSet) Slice() []peer.ID

func (ps *PeerSet) String() string

type Profile ¶ added in v0.0.9

type Profile interface {
	//
	Insert(req *Request, resp *Response)
	//
	Less(that Profile, req *Request) bool
}

Profile stores query profiles

type ProfileFactory ¶ added in v0.0.9

type ProfileFactory func() Profile

ProfileFactory generates a Profile

type PubOpt ¶

type PubOpt func(*PubOpts) error

PubOpt is optional options in PubSubCollector.Publish

func WithFinalRespHandler(handler FinalRespHandler) PubOpt

func WithRequestContext(ctx context.Context) PubOpt

type PubOpts ¶

type PubOpts struct {
	RequestContext  context.Context
	FinalRespHandle FinalRespHandler
}

PubOpts is the aggregated options

func NewPublishOptions(opts []PubOpt) (out *PubOpts, err error)

type PubReq ¶ added in v0.0.9

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

type PubSubCollector ¶

type PubSubCollector interface {

	// Join the overlay network defined by topic.
	// Register RequestHandle and ResponseHandle in opts.
	Join(topic string, opts ...JoinOpt) error

	// Publish a serialized request. Request should be encasulated in data argument.
	Publish(topic string, data []byte, opts ...PubOpt) error

	// Leave the overlay
	Leave(topic string) error

	io.Closer
}

PubSubCollector is a group communication module on topic-based overlay network. It helps to dispatch request, and wait for corresponding responses. In relay mode, PubSubCollector can also help to reduce the response.

func NewCollector(h host.Host, opts ...InitOpt) (PubSubCollector, error)

type PubSubFactory ¶

type PubSubFactory func(h host.Host) (*pubsub.PubSub, error)

PubSubFactory initializes a pubsub in Topics

type RelayPubSubCollector ¶

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

RelayPubSubCollector .

func NewRelayPubSubCollector(h host.Host, options ...InitOpt) (r *RelayPubSubCollector, err error)

func (r *RelayPubSubCollector) Close() (err error)

func (r *RelayPubSubCollector) Join(topic string, options ...JoinOpt) (err error)

func (r *RelayPubSubCollector) Leave(topic string) (err error)

func (r *RelayPubSubCollector) Publish(topic string, data []byte, opts ...PubOpt) (err error)

type ReqIDFn ¶

type ReqIDFn func(*Request) RequestID

ReqIDFn is used to generate id for each request

type Request ¶

type Request = pb.Request

Request type alias

type RequestHandler ¶

type RequestHandler func(ctx context.Context, req *Request) *Intermediate

RequestHandler is the callback function when receiving a request. It will be called in every node joined the network. The return value will be sent to the root (directly or relayedly).

type RequestID ¶

type RequestID = pb.RequestID

RequestID type alias

func DefaultReqIDFn(rq *Request) RequestID

type RequestResult ¶ added in v0.0.12

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

type Response ¶

type Response = pb.Response

Response type alias

type TopicHandle ¶

type TopicHandle func(topic string, msg *Message)

TopicHandle is the handle function of subscription. WARNING: DO NOT change msg, if a msg includes multiple topics, they share a message.

type TopicHub ¶ added in v0.0.9

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

func NewTopicHub(tw TopicWires) *TopicHub

func (th *TopicHub) Close() error

func (th *TopicHub) HandlePeerDown(p peer.ID, topic string)

func (th *TopicHub) HandlePeerUp(p peer.ID, topic string)

func (th *TopicHub) Join(topic string) error

func (th *TopicHub) JoinWithNotifiee(topic string, wn WireListener) error

func (th *TopicHub) Leave(topic string) error

func (th *TopicHub) Wires(topic string) Wires

type TopicMsgHandler ¶ added in v0.0.9

type TopicMsgHandler = pubsub.TopicMsgHandler

TopicMsgHandler .

type TopicOpt ¶

type TopicOpt func(*AsyncPubSub) error

TopicOpt is the option in NewTopics

func WithCustomPubSubFactory(psubFact PubSubFactory) TopicOpt

func WithSelfNotif(enable bool) TopicOpt

type TopicWireListener ¶ added in v0.0.9

type TopicWireListener = pubsub.TopicWireListener

TopicWireListener .

type TopicWires ¶ added in v0.0.9

type TopicWires interface {
	ID() peer.ID
	Join(topic string) error
	Leave(topic string) error
	Topics() []string
	Neighbors(topic string) []peer.ID
	SetListener(twn TopicWireListener)
	SendMsg(topic string, to peer.ID, data []byte) error
	SetTopicMsgHandler(th TopicMsgHandler)
	io.Closer
}

type WireListener ¶ added in v0.0.9

type WireListener interface {
	HandlePeerUp(p peer.ID)
	HandlePeerDown(p peer.ID)
}

type Wires ¶ added in v0.0.9

type Wires interface {
	ID() peer.ID
	Neighbors() []peer.ID
	SetListener(wn WireListener)
	SendMsg(to peer.ID, data []byte) error
	SetMsgHandler(h MsgHandler)
}