import { ILogger } from "@node-lightning/logger" ; import { Readable } from "stream" ; import { InitFeatureFlags } from "../flags/InitFeatureFlags" ; import { IWireMessage } from "../messages/IWireMessage" ; import { MessageType } from "../MessageType" ; import { IPeer , Peer } from "../Peer" ; import { PeerState } from "../PeerState" ; import { GossipError , GossipErrorCode } from "./GossipError" ; import { GossipFilter } from "./GossipFilter" ; import { GossipQueriesReceiver } from "./GossipQueriesReceiver" ; import { GossipQueriesSync } from "./GossipQueriesSync" ; export enum ReadState { Ready , Reading , Blocked } /**
 * Implements a decorator for the standard Peer adding functionality for Gossip
 * related activities. Messages from the GossipPeer occur after being pushed
 * through a GossipFilter to validate they are aokay.
 */ export class GossipPeer extends Readable implements IPeer { public readonly logger : ILogger ; private _readState : ReadState = ReadState . Ready ; private _receiver : GossipQueriesReceiver ; private _syncRangeTask : GossipQueriesSync ; param peer param filter param logger constructor ( readonly peer , readonly filter , logger ) { super ( { objectMode : true , highWaterMark : peer . highWaterMark } ) ; if ( peer . state !== PeerState . Ready ) { throw new GossipError ( GossipErrorCode . PeerNotReady ) ; } this . logger = logger . sub ( "gspeer" , peer . id ) ; this . peer . on ( "readable" , this . _onPeerReadable . bind ( this ) ) ; this . peer . on ( "close" , ( ) => this . emit ( "close" ) ) ; if ( this . gossipQueries ) { this . _receiver = new GossipQueriesReceiver ( this . peer . localFeatures [ 0 ] , this . peer , this . logger ) ; } } /**
     * Returns true if the gossip_queries feature has been negotiated with the
     * remote peer.
     */ public get gossipQueries ( ) : boolean { return ( this . peer . remoteFeatures . isSet ( InitFeatureFlags . gossipQueriesOptional ) || this . peer . remoteFeatures . isSet ( InitFeatureFlags . gossipQueriesRequired ) ) ; } param firstBlock param numBlocks public async syncRange ( firstBlock ? , numBlocks ? ) { if ( this . gossipQueries ) { const chainHash = this . peer . localChains [ 0 ] ; this . _syncRangeTask = new GossipQueriesSync ( chainHash , this , this . logger ) ; await this . _syncRangeTask . queryRange ( firstBlock , numBlocks ) ; this . _syncRangeTask = undefined ; return true ; } return false ; } /**
     * Enables the receipt of rumor mongered messages.
     */ public enableGossip ( ) { if ( this . gossipQueries ) { this . _receiver . activate ( ) ; } } /**
     * Disables the receipt of rumor mongered messages.
     */ public disableGossip ( ) { if ( this . gossipQueries ) { this . _receiver . deactivate ( ) ; } } param buf public send ( buf ) { this . peer . send ( buf ) ; } param msg public sendMessage ( msg ) { this . peer . sendMessage ( msg ) ; } /**
     * Disconnects the peer
     */ public disconnect ( ) { this . peer . disconnect ( ) ; } returns protected async _onPeerReadable ( ) { while ( this . _readState === ReadState . Ready ) { this . _readState = ReadState . Reading ; const msg = this . peer . read ( ) as IWireMessage ; if ( ! msg ) { this . _readState = ReadState . Ready ; return ; } const filteredMsgs = await this . _onPeerMessage ( msg ) ; this . _readState = ReadState . Ready ; for ( const filteredMsg of filteredMsgs ) { if ( ! this . push ( filteredMsg ) ) { this . _readState = ReadState . Blocked ; } } } } param msg protected async _onPeerMessage ( msg ) { if ( this . _syncRangeTask ) { this . _syncRangeTask . handleWireMessage ( msg ) ; } if ( msg . type === MessageType . ChannelAnnouncement || msg . type === MessageType . ChannelUpdate || msg . type === MessageType . NodeAnnouncement ) { try { const result = await this . filter . validateMessage ( msg ) ; if ( result . isOk ) { return result . value ; } else { this . emit ( "gossip_error" , result . error ) ; } } catch ( err ) { this . disconnect ( ) ; this . emit ( "error" , err ) ; return [ ] ; } } return [ msg ] ; } public _read ( ) { if ( this . _readState === ReadState . Blocked ) { this . _readState = ReadState . Ready ; } setImmediate ( ( ) => this . _onPeerReadable ( ) ) ; } }