### Question:
Zmq { public static Sink<ZMsg, NotUsed> publishServerSink(String addresses) { return Flow.fromGraph(new ZmqPublishStage(true, addresses)).to(Sink.ignore()); }  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 shouldServerPublishMessageWhenSourceElementAvailable() throws InterruptedException { final int port = findFreePort(); final String addr = "tcp: Source<ZMsg, TestPublisher.Probe<ZMsg>> testSource = TestSource.probe(system); TestPublisher.Probe<ZMsg> probe = Zmq.publishServerSink(addr).runWith(testSource, materializer); probe.ensureSubscription(); Thread.sleep(200); final ZMQ.Context context = ZMQ.context(1); final ZMQ.Socket socket = context.socket(ZMQ.SUB); socket.connect(addr); socket.subscribe(ZMQ.SUBSCRIPTION_ALL); Thread.sleep(200); probe.sendNext(ZMsg.newStringMsg("test")); Thread.sleep(200); String s = socket.recvStr(ZMQ.DONTWAIT); Assert.assertEquals(s, "test"); socket.close(); context.term(); }