libp2p/protocols/pubsub/pubsubpeer

Search:
Group by:
Source   Edit  

Types

CustomPeerSelectionProc = proc (allPeers: HashSet[PubSubPeer];
                                directPeers: HashSet[PubSubPeer];
                                meshPeers: HashSet[PubSubPeer];
                                fanoutPeers: HashSet[PubSubPeer]): HashSet[
    PubSubPeer] {....gcsafe, raises: [].}
Source   Edit   
CustomStreamCallbacks = object
  customStreamCreationCB*: CustomStreamCreationProc
  customPeerSelectionCB*: CustomPeerSelectionProc
Source   Edit   
CustomStreamCreationProc = proc (destAddr: Opt[MultiAddress];
                                 destPeerId: PeerId; codec: string): Stream {.
    ...gcsafe, raises: [].}
Source   Edit   
DropStream = proc (peer: PubSubPeer) {....gcsafe, raises: [].}
Source   Edit   
GetStream = proc (): Future[Stream] {.async: (
    ...raises: [CancelledError, GetStreamDialError]).}
Source   Edit   
GetStreamDialError = object of CatchableError
Source   Edit   
MessagePriority {.pure.} = enum
  High, ## Protocol-critical messages: subscriptions, GRAFT, PRUNE, IHAVE,
         ## IDontWant, control responses. Sent immediately. If max is exceeded
         ## we drop the peer.
  Medium, ## Locally published messages. Sent before low priority traffic.
           ## They are dropped if exceeded max. Peer is not disconnected.
  Low                        ## Relayed messages and IWANT replies
                             ## They are dropped if exceeded max. Peer is not disconnected.
Source   Edit   
OnEvent = proc (peer: PubSubPeer; event: PubSubPeerEvent) {....gcsafe, raises: [].}
Source   Edit   
PeerRateLimitError = object of CatchableError
Source   Edit   
PubSubObserver = ref object
  onRecv*: proc (peer: PubSubPeer; msgs: var RPCMsg) {....gcsafe, raises: [].}
  onSend*: proc (peer: PubSubPeer; msgs: var RPCMsg) {....gcsafe, raises: [].}
  onValidated*: proc (peer: PubSubPeer; msg: Message; msgId: MessageId) {.
      ...gcsafe, raises: [].}
Source   Edit   
PubSubPeer = ref object of RootObj
  getStream*: GetStream
  onEvent*: OnEvent
  codec*: string
  sendStream*: Stream
  address*: Opt[MultiAddress]
  peerId*: PeerId
  handler*: RPCHandler
  observers*: ref seq[PubSubObserver]
  score*: float64
  subscribedTopics*: int
  sentIHaves*: Deque[HashSet[MessageId]]
  iDontWants*: Deque[HashSet[SaltedId]] ## IDONTWANT contains unvalidated message id:s which may be long and/or
                                        ## expensive to look up, so we apply the same salting to them as during
                                        ## unvalidated message processing
  iHaveBudget*: int
  appScore*: float64
  slowPeerPenalty*: float64
  behaviourPenalty*: float64
  overheadRateLimitOpt*: Opt[TokenBucket]
  maxHighPriorityQueueLen*: int
  maxMediumPriorityQueueLen*: int
  maxLowPriorityQueueLen*: int
  customStreamCallbacks*: Opt[CustomStreamCallbacks]
Source   Edit   
PubSubPeerEventKind {.pure.} = enum
  StreamOpened, StreamClosed, DisconnectionRequested
Source   Edit   
QueueAction = object
  priority*: MessagePriority
  send*: bool
  slowPeerPenaltyDelta*: float64
Source   Edit   
RPCHandler = proc (peer: PubSubPeer; data: sink seq[byte]): Future[void] {.
    async: (...raises: [CancelledError]).}
Source   Edit   
RpcMessageQueue = ref object
Source   Edit  

Procs

proc `$`(p: PubSubPeer): string {....raises: [], tags: [], forbids: [].}
Source   Edit   
func `==`(a, b: PubSubPeer): bool {....raises: [], tags: [], forbids: [].}
Source   Edit   
proc canAskIWant(p: PubSubPeer; msgId: MessageId): bool {....raises: [], tags: [],
    forbids: [].}
Source   Edit   
proc connect(p: PubSubPeer) {....raises: [], tags: [RootEffect], forbids: [].}
Source   Edit   
proc connected(p: PubSubPeer): bool {....raises: [], tags: [], forbids: [].}
Source   Edit   
func determineQueueAction(priority: MessagePriority;
                          rpcmessagequeue: RpcMessageQueue;
    maxHighPriorityQueueLen, maxMediumPriorityQueueLen, maxLowPriorityQueueLen: int): QueueAction {.
    ...raises: [], tags: [], forbids: [].}
Source   Edit   
func determineQueueAction(priority: MessagePriority; sendPriorityQueueLen,
    mediumPriorityQueueLen, lowPriorityQueueLen: int; maxHighPriorityQueueLen,
    maxMediumPriorityQueueLen, maxLowPriorityQueueLen: int): QueueAction {.
    ...raises: [], tags: [], forbids: [].}
Source   Edit   
proc getAgent(peer: PubSubPeer): string {....raises: [], tags: [], forbids: [].}
Source   Edit   
func hash(p: PubSubPeer): Hash {....raises: [], tags: [], forbids: [].}
Source   Edit   
proc hasObservers(p: PubSubPeer): bool {....raises: [], tags: [], forbids: [].}
Source   Edit   
proc hasSendStream(p: PubSubPeer): bool {....raises: [], tags: [], forbids: [].}
Source   Edit   
proc new(T: typedesc[PubSubPeer]; peerId: PeerId; getStream: GetStream;
         onEvent: OnEvent; codec: string; maxMessageSize: int;
         maxHighPriorityQueueLen: int = DefaultMaxHighPriorityQueueLen;
         maxMediumPriorityQueueLen: int = DefaultMaxMediumPriorityQueueLen;
         maxLowPriorityQueueLen: int = DefaultMaxLowPriorityQueueLen;
         overheadRateLimitOpt: Opt[TokenBucket] = Opt.none(TokenBucket);
    customStreamCallbacks: Opt[CustomStreamCallbacks] = Opt.none(
    CustomStreamCallbacks)): T:type {....raises: [].}
Source   Edit   
func outbound(p: PubSubPeer): bool {....raises: [], tags: [], forbids: [].}
Source   Edit   
proc recvObservers(p: PubSubPeer; msg: var RPCMsg) {....raises: [],
    tags: [RootEffect], forbids: [].}
Source   Edit   
proc runHandleLoop(p: PubSubPeer; stream: Stream): InternalRaisesFuture[void,
    (CancelledError,)] {....stackTrace: false, raises: [], gcsafe, raises: [],
                         tags: [RootEffect], forbids: [].}
Source   Edit   
proc send(p: PubSubPeer; msg: RPCMsg; anonymize: bool;
          priority: MessagePriority; useCustomStream: bool = false) {.
    ...raises: [], raises: [], tags: [RootEffect, WriteIOEffect], forbids: [].}

Asynchronously sends an RPCMsg to a specified PubSubPeer with an option for anonymization.

Parameters:

  • p: The PubSubPeer instance to which the message is to be sent.
  • msg: The RPCMsg instance representing the message to be sent.
  • anonymize: A boolean flag indicating whether the message should be sent with anonymization.
  • priority: The message priority level (High, Medium, or Low). High priority messages are sent immediately, medium and low priority messages are queued and sent only after all high priority messages have been sent.
Source   Edit   
proc sendEncoded(p: PubSubPeer; msg: sink seq[byte]; priority: MessagePriority;
                 useCustomStream: bool = false): Future[void] {....raises: [],
    tags: [RootEffect, WriteIOEffect], forbids: [].}

Asynchronously sends an encoded message to a specified PubSubPeer according to its priority.

Parameters:

  • p: The PubSubPeer instance to which the message is to be sent.
  • msg: The message to be sent, encoded as a sequence of bytes (seq[byte]).
  • priority:
    • High or any priority when all queues are empty: sent immediately
    • Medium: queued in mediumPriorityQueue. Dropped when full.
    • Low: queued in lowPriorityQueue. Dropped when full.
  • useCustomStream: boolean used to indicate if a custom stream is going to be used for sending this message

Low and medium priority messages are queued and sent only after all high priority messages have been sent.

Source   Edit   
func shortLog(p: PubSubPeer): string {....raises: [], tags: [], forbids: [].}
Source   Edit   
proc stopTasks(p: PubSubPeer) {....raises: [], tags: [RootEffect, WriteIOEffect],
                                forbids: [].}
Peer-wide teardown: cancels the connector loop, in-flight sends, and the non-high-priority sender task; clears its priority queues. Source   Edit   
proc trackSend(p: PubSubPeer; fut: Future[void]) {....raises: [], tags: [],
    forbids: [].}
Take ownership of an in-flight send future on this peer. Source   Edit   
proc validatedObservers(p: PubSubPeer; msg: Message; msgId: MessageId) {.
    ...raises: [], tags: [RootEffect], forbids: [].}
Source   Edit  

Templates

template chroniclesFormatItIMPL(it: PubSubPeer): auto {..}
Source   Edit  

Exports

hash, shortLog, init, random, supportsPacked, $, write, PeerId, ==, getField, <=, toCidString, shortLog, computeFieldSize, shortLog, capLen, init, validate, init, supportsPacked, exclIfIt, match, <, chroniclesFormatItIMPL, write, getPubKey, maxInlineKeyLength, readFieldInto, withValue, writeField, >=, withValue, computeFieldSize, shortLog, writeField, toBytes, filterIt, match, readFieldInto, toChunks, toCid, getBytes, init, init, take, toOpt, toOpt, init, withValue, hasPublicKey, chroniclesFormatItIMPL, random, >, hex, len, toOpt, cmp, extractPublicKey, DNS_OR_IP, shortLog, init, random, UDP_DNS, $, MaError, libp2p_network_bytes, writeField, ephemeral, replaceIp, MaPatternOp, ==, WS_DNS, initVBuffer, <=, MaxSizeError, decode, ==, shortLog, filterIt, ECDHEScheme, shortLog, hash, supportsPacked, writeField, init, closed, [], shuffle, random, init, Secret, peekVarint, payload, $, validate, mapOr, chroniclesFormatItIMPL, init, Memory, bytes, publicRoutableAddressPolicy, getPubKey, maxInlineKeyLength, TranscoderDNS, QUIC, writeField, [], init, ephemeral, withValue, PeerInfoObserver, getIp, generateBytes, WS_IP, write, MAKind, decode, getBytes, write, decode, init, noPrivateAddressPolicy, getRepeatedField, generate, matchPartial, getBytes, len, areAddrsConsistent, init, readLp, HTTP, initVBuffer, atEof, LPStreamTrackerName, removeObserver, peekSeq, update, TCP_IP4, toOpt, join, fullAddrs, payloadType, LPStreamError, init, IP, protoAddress, QUIC_V1_IP, newRng, ==, shortLog, withValue, checkValid, getPublicKey, chroniclesFormatItIMPL, libp2p_multiaddress_exts, WebSockets, readFieldInto, getPart, CircuitRelay, extractPublicKey, mapAnd, new, maErr, computeFieldSize, UDP_IP, newLPStreamEOFError, init, random, ==, DNSADDR, AddressMapper, write, notifyObservers, WSS, QUIC_V1, TCP_IP, close, init, closeImpl, toCidString, readSeq, toBytes, TCP_DNS, closeWrite, shortLog, wasResetLocally, encode, UDP_IP4, len, capLen, init, protoName, protoArgument, toRawBytes, supportsPacked, $, writeField, TCP_IP6, init, DNS, WSS_DNS, match, writeLPVarint, CryptoResult, LPStream, parseFullAddress, WebRTCDirect, PeerAddressPolicy, init, $, UDP, PeerInfo, TranscoderOnion, StreamTracker, defaultAddressPolicy, init, LPStreamResetError, writeArray, LPStreamIncompleteError, PeerId, >, [], LPStreamEOFError, IP6, shortLog, reset, initVBuffer, PKScheme, random, stretchKeys, accepts, addObserver, Connection, write, init, generate, TCP, PrivateKey, readFieldInto, write, init, RawConn, &=, PeerInfoError, newBearSslRng, verify, newLPStreamClosedError, random, TranscoderMemory, toString, TranscoderIP4, init, toException, init, goffset, getRawBytes, pick, KeyPair, init, high, cmp, init, DefaultConnectionTimeout, Signature, chroniclesFormatItIMPL, contains, hash, exclIfIt, computeFieldSize, computeFieldSize, supportsPacked, stripPeerId, toBytes, DigestSheme, IPFS, getField, shortLog, readOnce, init, resetImpl, isEmpty, shortLog, data, chroniclesFormatItIMPL, hex, protocols, chroniclesFormatItIMPL, computeFieldSize, HTTPS, UNIX, LPError, supportsPacked, readFieldInto, TLS_WS, sign, encode, Rng, getProtocolArgument, supported, MaResult, init, supportsPacked, toBytes, QUIC_V1_IP4, writeSeq, finish, &, DNS6, expandAddrs, writeField, decode, IpTransportProtocol, key, IP4, TranscoderPort, getBytes, readFieldInto, DNSANY, >=, QUIC_V1_DNS, payloadDomain, $, MaPattern, withValue, MuxedStream, init, LPStreamLimitError, TranscoderOnion3, SignedPayload, mac, toBytes, protoArgument, newLPStreamLimitError, keyOpenArray, match, WebSockets_DNS, SignedPeerRecord, $, isEmpty, writeLp, toCid, init, newLPStreamIncompleteError, MAProtocol, newLPStreamResetError, initStream, init, getField, DNS4, readFieldInto, getWrapped, selectBest, supportsPacked, P2PPattern, Envelope, hasPublicKey, toException, init, Stream, isEnough, Reliable, addrs, hex, closeImpl, getBytes, ivOpenArray, WS, init, encode, take, init, init, match, readArray, filterAddrs, <, iv, readLine, Eof, CryptoError, computeFieldSize, append, hash, init, parseFullAddress, WSS_IP, LPStreamConnDownError, peekArray, PeerRecord, SupportedSchemes, generate, QUIC_DNS, Unreliable, SupportedSchemesInt, random, checkFutures, supportsPacked, macOpenArray, init, checkAddresses, validate, protoCode, mapEq, libp2p_pki_schemes, Onion3, TcpOnion3, toFullAddress, pickOne, concat, TranscoderP2P, AddressInfo, ConnectionTrackerName, TranscoderIP6, new, write, EnvelopeError, computeFieldSize, encode, len, newLPStreamRemoteClosedError, TranscoderIP6Zone, MaPatResult, Direction, init, InvalidVarintError, shortLog, items, init, write, write, init, UTP, readVarint, verify, readFieldInto, TimeoutHandler, QUIC_IP, init, PublicKey, WebSockets_IP, VBuffer, writeLp, writePBVarint, toChunks, readExactly, newLPStreamConnDownError, init, initStream, MultiAddress, supportsPacked, readVarint, toOpt, closeWithEOF, RsaDefaultKeySize, orError, LPStreamRemoteClosedError, <, MaInvalidAddress, generate, LPStreamClosedError, writeField, write, getOrder, TranscoderUnix, toOpt, writeVarint

Made with Nim. Generated: 2026-07-14 05:50:27 UTC