DropConn = proc (peer: PubSubPeer) {....gcsafe, raises: [].}

GetConn = proc (): Future[Connection] {....gcsafe, raises: [].}

OnEvent = proc (peer: PubSubPeer; event: PubSubPeerEvent) {....gcsafe, raises: [].}

PeerRateLimitError = object of CatchableError

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: [].}

PubSubPeer = ref object of RootObj
  getConn*: GetConn
  onEvent*: OnEvent
  codec*: string
  sendConn*: Connection
  connectedFut: Future[void]
  address*: Option[MultiAddress]
  peerId*: PeerId
  handler*: RPCHandler
  observers*: ref seq[PubSubObserver]
  score*: float64
  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
  pingBudget*: int
  maxMessageSize: int
  appScore*: float64
  behaviourPenalty*: float64
  overheadRateLimitOpt*: Opt[TokenBucket]
  rpcmessagequeue: RpcMessageQueue
  maxNumElementsInNonPriorityQueue*: int
  disconnected: bool

PubSubPeerEvent = object
  kind*: PubSubPeerEventKind

PubSubPeerEventKind {.pure.} = enum
  StreamOpened, StreamClosed, DisconnectionRequested

RPCHandler = proc (peer: PubSubPeer; data: seq[byte]): Future[void] {....gcsafe,
    raises: [].}

RpcMessageQueue = ref object
  sendPriorityQueue: Deque[Future[void]]
  nonPriorityQueue: AsyncQueue[seq[byte]]
  sendNonPriorityTask: Future[void]

DefaultMaxNumElementsInNonPriorityQueue = 1024

func `==`(a, b: PubSubPeer): bool {....raises: [], tags: [].}

proc canAskIWant(p: PubSubPeer; msgId: MessageId): bool {....raises: [], tags: [].}

proc connect(p: PubSubPeer) {....raises: [], tags: [RootEffect].}

proc connected(p: PubSubPeer): bool {....raises: [], tags: [].}

proc getAgent(peer: PubSubPeer): string {....raises: [], tags: [].}

proc handle(p: PubSubPeer; conn: Connection): Future[void] {.stackTrace: false,
    ...raises: [], gcsafe, raises: [], raises: [], tags: [RootEffect].}

func hash(p: PubSubPeer): Hash {....raises: [], tags: [].}

proc hasObservers(p: PubSubPeer): bool {....raises: [], tags: [].}

proc hasSendConn(p: PubSubPeer): bool {....raises: [], tags: [].}

proc new(T: typedesc[PubSubPeer]; peerId: PeerId; getConn: GetConn;
         onEvent: OnEvent; codec: string; maxMessageSize: int;
    maxNumElementsInNonPriorityQueue: int = DefaultMaxNumElementsInNonPriorityQueue;
         overheadRateLimitOpt: Opt[TokenBucket] = Opt.none(TokenBucket)): T:type {.
    ...raises: [].}

func outbound(p: PubSubPeer): bool {....raises: [], tags: [].}

proc recvObservers(p: PubSubPeer; msg: var RPCMsg) {....raises: [],
    tags: [RootEffect].}

proc send(p: PubSubPeer; msg: RPCMsg; anonymize: bool; isHighPriority: bool) {.
    ...raises: [], raises: [], tags: [RootEffect, WriteIOEffect, TimeEffect].}

proc sendEncoded(p: PubSubPeer; msg: seq[byte]; isHighPriority: bool): Future[
    void] {....raises: [], tags: [WriteIOEffect, TimeEffect, RootEffect].}

func shortLog(p: PubSubPeer): string {....raises: [], tags: [].}

proc stopSendNonPriorityTask(p: PubSubPeer) {....raises: [],
    tags: [WriteIOEffect, TimeEffect, RootEffect].}

proc validatedObservers(p: PubSubPeer; msg: Message; msgId: MessageId) {.
    ...raises: [], tags: [RootEffect].}

template formatItIMPL(it: PubSubPeer): auto