import * as zmq from "../../src" ; import { assert } from "chai" ; import { testProtos , uniqAddress } from "./helpers" ; for ( const proto of testProtos ( "tcp" , "ipc" , "inproc" ) ) { describe ( `socket with ${ proto } pub/sub` , function ( ) { let pub ; let sub ; beforeEach ( function ( ) { pub = new zmq . Publisher ( ) ; sub = new zmq . Subscriber ( ) ; } ) ; afterEach ( function ( ) { pub . close ( ) ; sub . close ( ) ; global . gc ?. ( ) ; } ) ; describe ( "send/receive" , function ( ) { it ( "should deliver messages" , async function ( ) { const address = uniqAddress ( proto ) ; const messages = [ "foo" , "bar" , "baz" , "qux" ] ; const received = [ ] ; sub . subscribe ( ) ; await sub . bind ( address ) ; await pub . connect ( address ) ; const send = async ( ) => { await new Promise ( resolve => setTimeout ( resolve , 25 ) ) ; for ( const msg of messages ) { await pub . send ( msg ) ; } } ; const receive = async ( ) => { for await ( const [ msg ] of sub ) { assert . instanceOf ( msg , Buffer ) ; received . push ( msg . toString ( ) ) ; if ( received . length === messages . length ) break ; } } ; await Promise . all ( [ send ( ) , receive ( ) ] ) ; assert . deepEqual ( received , messages ) ; } ) ; } ) ; describe ( "subscribe/unsubscribe" , function ( ) { it ( "should filter messages" , async function ( ) { const address = uniqAddress ( proto ) ; const messages = [ "foo" , "bar" , "baz" , "qux" ] ; const received = [ ] ; sub . subscribe ( "fo" , "ba" , "qu" ) ; sub . unsubscribe ( "fo" , "qu" ) ; await sub . bind ( address ) ; await pub . connect ( address ) ; const send = async ( ) => { await new Promise ( resolve => setTimeout ( resolve , 25 ) ) ; for ( const msg of messages ) { await pub . send ( msg ) ; } } ; const receive = async ( ) => { for await ( const [ msg ] of sub ) { assert . instanceOf ( msg , Buffer ) ; received . push ( msg . toString ( ) ) ; if ( received . length === 2 ) break ; } } ; await Promise . all ( [ send ( ) , receive ( ) ] ) ; assert . deepEqual ( received , [ "bar" , "baz" ] ) ; } ) ; } ) ; } ) ; }