### Question:
Zmq { public static Source<ZMsg, NotUsed> pullServerSource(String addresses) { return Source.fromGraph(new ZmqPullStage(true, addresses)); }  static Source<ZMsg, NotUsed> subscribeClientSource(String addresses); static Source<ZMsg, NotUsed> subscribeClientSource(String addresses, String subscription); static Source<ZMsg, NotUsed> pullServerSource(String addresses); static Source<ZMsg, NotUsed> pullClientSource(String addresses); static Sink<ZMsg, NotUsed> publishServerSink(String addresses); static Flow<ZMsg, ZMsg, NotUsed> publishServerFlow(String addresses); static Sink<ZMsg, NotUsed> pushServerSink(String addresses); static Sink<ZMsg, NotUsed> pushClientSink(String addresses); static Flow<ZMsg, ZMsg, NotUsed> pushServerFlow(String addresses); static Flow<ZMsg, ZMsg, NotUsed> pushClientFlow(String addresses);  }### Answer:
@Test public void shouldEmitElementWhenServerPullReceiveMessage() throws InterruptedException { final int port = findFreePort(); final String addr = "tcp: final ZMQ.Context context = ZMQ.context(1); final ZMQ.Socket socket = context.socket(ZMQ.PUSH); socket.connect(addr); Thread.sleep(200); Sink<ZMsg, TestSubscriber.Probe<ZMsg>> testSink = TestSink.probe(system); TestSubscriber.Probe<ZMsg> probe = Zmq.pullServerSource(addr).runWith(testSink, materializer); probe.ensureSubscription(); Thread.sleep(200); socket.send("test"); Thread.sleep(200); ZMsg zMsg = probe.requestNext(); Assert.assertEquals("test", zMsg.popString()); socket.close(); context.term(); }