import { Message , EventMessage , RequestMessage , ResponseMessage } from './message' ; import { ProtocolMeta as Meta , Channel } from '@uniqys/p2p-network' ; import { Duplex } from 'pull-stream' ; import { EventEmitter } from 'events' ; import PeerInfo from 'peer-info' ; export { Message } ; export interface ProtocolHandler { handshake ( protocol , incoming ) : void ; hello ( message , protocol ) : void ; newTransaction ( message , protocol ) : void ; newBlock ( message , protocol ) : void ; newBlockHeight ( message , protocol ) : void ; newConsensusMessage ( message , protocol ) : void ; getConsentedHeader ( message , protocol ) : Promise < Message . ConsentedHeader > ; getHeaders ( message , protocol ) : Promise < Message . Headers > ; getBodies ( message , protocol ) : Promise < Message . Bodies > ; } export class Protocol { private receiver ? : ( msg : ResponseMessage ) => void ; private readonly event = new EventEmitter ( ) ; constructor ( public readonly peerId , private readonly channel , private readonly handler ) { this . channel . onError ( err => this . event . emit ( 'error' , err ) ) ; this . channel . onEnd ( ( ) => this . event . emit ( 'end' ) ) ; } public start ( ) { this . channel . onMessage ( msg => { if ( msg . type === Message . Type . Event ) { this . handleEvent ( msg ) ; } if ( msg . type === Message . Type . Request ) { this . handleRequest ( msg ) ; } if ( msg . type === Message . Type . Response ) { this . handleResponse ( msg ) ; } } ) ; } public end ( ) { this . channel . end ( ) ; } public onError ( listener ) { this . event . on ( 'error' , listener ) ; } public onEnd ( listener ) { this . event . on ( 'end' , listener ) ; } public sendHello ( message ) { return this . channel . sendMessage ( message ) ; } public sendNewTransaction ( message ) { return this . channel . sendMessage ( message ) ; } public sendNewBlock ( message ) { return this . channel . sendMessage ( message ) ; } public sendNewBlockHeight ( message ) { return this . channel . sendMessage ( message ) ; } public sendNewConsensusMessage ( message ) { return this . channel . sendMessage ( message ) ; } public fetchConsentedHeader ( message ) { return this . fetch ( message , Message . ConsentedHeader ) ; } public fetchHeaders ( message ) { return this . fetch ( message , Message . Headers ) ; } public fetchBodies ( message ) { return this . fetch ( message , Message . Bodies ) ; } private handleEvent ( msg ) { if ( msg instanceof Message . Hello ) return this . handler . hello ( msg , this ) ; if ( msg instanceof Message . NewTransaction ) return this . handler . newTransaction ( msg , this ) ; if ( msg instanceof Message . NewBlock ) return this . handler . newBlock ( msg , this ) ; if ( msg instanceof Message . NewBlockHeight ) return this . handler . newBlockHeight ( msg , this ) ; if ( msg instanceof Message . NewConsensusMessage ) return this . handler . newConsensusMessage ( msg , this ) ; this . unexpected ( msg ) ; } private handleRequest ( msg ) { if ( msg instanceof Message . GetConsentedHeader ) { this . handler . getConsentedHeader ( msg , this ) . then ( msg => this . channel . sendMessage ( msg ) , err => this . event . emit ( 'error' , err ) ) ; return ; } if ( msg instanceof Message . GetHeaders ) { this . handler . getHeaders ( msg , this ) . then ( msg => this . channel . sendMessage ( msg ) , err => this . event . emit ( 'error' , err ) ) ; return ; } if ( msg instanceof Message . GetBodies ) { this . handler . getBodies ( msg , this ) . then ( msg => this . channel . sendMessage ( msg ) , err => this . event . emit ( 'error' , err ) ) ; return ; } this . unexpected ( msg ) ; } private handleResponse ( msg ) { if ( this . receiver ) { this . receiver ( msg ) ; } else { this . event . emit ( 'error' , new Error ( 'unexpected response' ) ) ; } } private async fetch < T > ( message , ctor ) { const response = new Promise < T > ( ( resolve , reject ) => { this . receiver = ( msg => { this . receiver = undefined ; if ( msg instanceof ctor ) { resolve ( msg ) ; } else { const err = new Error ( 'unexpected response' ) ; this . event . emit ( 'error' , err ) ; reject ( err ) ; } } ) ; } ) ; await this . channel . sendMessage ( message ) ; return response ; } private unexpected ( msg ) { this . event . emit ( 'error' , new Error ( `unexpected message ${ msg }` ) ) ; } } export class ProtocolMeta implements Meta { public protocol = 'uniqys/v1' ; constructor ( private readonly handler ) { } handshake ( info , conn , incoming ) { const protocol = new Protocol ( info . id . toB58String ( ) , new Channel ( conn , Message . deserialize , Message . serialize ) , this . handler ) ; this . handler . handshake ( protocol , incoming ) ; } }