  @Test
  public void shouldServerPublishMessageWhenSourceElementAvailable() throws InterruptedException {

    //Given
    final int port = findFreePort();
    final String addr = "tcp://localhost:" + port;
   //When
    Source<ZMsg, TestPublisher.Probe<ZMsg>> testSource = TestSource.probe(system);

    TestPublisher.Probe<ZMsg> probe = Zmq.publishServerSink(addr).runWith(testSource, materializer);

    probe.ensureSubscription();
    Thread.sleep(200); //wait for published to bind

    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); //wait for subscriber to connect


    //Then
    probe.sendNext(ZMsg.newStringMsg("test"));
    Thread.sleep(200);

    String s = socket.recvStr(ZMQ.DONTWAIT);
    Assert.assertEquals(s, "test");

    socket.close();
    context.term();
  }