/**
 * Initializes common fixture.
 */
@BeforeEach
public void setUp() {

    connection = mock(ProtonConnection.class);
    vertx = mock(Vertx.class);
    eventBus = mock(EventBus.class);
    receiver = mock(ProtonReceiver.class);
    when(receiver.handler(any())).thenReturn(receiver);
    when(receiver.closeHandler(any())).thenReturn(receiver);
    when(receiver.setAutoAccept(any(Boolean.class))).thenReturn(receiver);
    when(receiver.setPrefetch(any(Integer.class))).thenReturn(receiver);
    when(receiver.setQoS(any(ProtonQoS.class))).thenReturn(receiver);

    when(vertx.eventBus()).thenReturn(eventBus);

    final ProtonSession session = mock(ProtonSession.class);
    when(session.getConnection()).thenReturn(connection);
    sender = mock(ProtonSender.class);
    when(sender.getName()).thenReturn("mocked sender");
    when(sender.isOpen()).thenReturn(Boolean.TRUE);
    when(sender.getSession()).thenReturn(session);
}
 

private void connectHandler(ProtonConnection connection) {
    connection.setContainer(containerId);
    connection.openHandler(conn -> {
        log.info("[{}]: Connection opened", containerId);
    }).closeHandler(conn -> {
        connection.close();
        connection.disconnect();
        log.info("[{}]: Connection closed", containerId);
    }).disconnectHandler(protonConnection -> {
        connection.disconnect();
        log.debug("Disconnected");
    }).open();

    connection.sessionOpenHandler(ProtonSession::open);
    connection.senderOpenHandler(sender -> senderOpenHandler(connection, sender));
    connection.receiverOpenHandler(receiver -> receiverOpenHandler(connection, receiver));
}
 

private ProtonSession getDefaultSession() {
  if (defaultSession == null) {
    defaultSession = createSession();
    defaultSession.closeHandler(result -> {
      String msg = "The connections default session closed unexpectedly";
      if (!result.succeeded()) {
        msg += ": ";
        msg += ": " + String.valueOf(result.cause());
      }
      Future<ProtonConnection> failure = Future.failedFuture(msg);
      Handler<AsyncResult<ProtonConnection>> connCloseHandler = closeHandler;
      if (connCloseHandler != null) {
        connCloseHandler.handle(failure);
      }
    });

    defaultSession.open();
    // Deliberately not flushing, the sender/receiver open
    // call will do that (if it doesn't happen otherwise).
  }
  return defaultSession;
}
 

/**
 * This method is called when an AMQP BEGIN frame is received from a remote client. This method sets the incoming
 * capacity in its BEGIN Frame to be communicated to the remote peer
 *
 */
private void handleSessionOpen(final ProtonConnection conn, final ProtonSession session) {
    log.debug("opening new session with client [container: {}, session window size: {}]",
            conn.getRemoteContainer(), getConfig().getMaxSessionWindowSize());
    session.setIncomingCapacity(getConfig().getMaxSessionWindowSize());
    session.open();
}
 

/**
 * Verify that a second response link to the same address is being accepted if the first is released before.
 */
@Test
public void testFreeSubscription() {

    final ProtonConnection con1 = mock(ProtonConnection.class);
    final ProtonSession session1 = mock(ProtonSession.class);
    when(session1.getConnection()).thenReturn(con1);
    final ProtonConnection con2 = mock(ProtonConnection.class);

    final ProtonSender sender1 = mock(ProtonSender.class);
    when(sender1.getSession()).thenReturn(session1);
    final ProtonSender sender2 = mock(ProtonSender.class);

    final RequestResponseEndpoint<ServiceConfigProperties> endpoint = getEndpoint(true);

    // WHEN a first sender attaches
    endpoint.onLinkAttach(con1, sender1, REPLY_RESOURCE);

    // THEN open has to be called
    verify(sender1).open();

    // WHEN the connection closed
    endpoint.onConnectionClosed(con1);

    // WHEN a new link is attached
    endpoint.onLinkAttach(con2, sender2, REPLY_RESOURCE);

    // THEN open has to be called
    verify(sender2).open();
}
 

/**
 * Handler for session closing from the remote
 *
 * @param session related Proton session closed
 */
private void processOpenSession(ProtonSession session) {

    session.closeHandler(ar -> {

        if (ar.succeeded()) {
            ar.result().close();
        }

    }).open();
}
 

@BeforeEach
public void setup() throws InterruptedException {
    vertx = Vertx.vertx();
    server = ProtonServer.create(vertx);
    CountDownLatch latch = new CountDownLatch(1);
    CompletableFuture<ProtonSender> futureSender = new CompletableFuture<>();
    server.connectHandler(conn -> {
        conn.closeHandler(c -> {
            conn.close();
            conn.disconnect();
        });
        conn.disconnectHandler(c -> {
            conn.disconnect();
        }).open();

        conn.sessionOpenHandler(ProtonSession::open);

        conn.receiverOpenHandler(receiver -> {
            System.out.println("Receiver open");
            receiver.setTarget(receiver.getRemoteTarget());
            receiver.handler((delivery, message) -> {
                Message response = Message.Factory.create();
                response.setAddress(message.getAddress());
                response.setBody(new AmqpValue(true));
                response.setCorrelationId(message.getCorrelationId());
                response.setReplyTo(message.getReplyTo());
                try {
                    futureSender.get().send(response);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            receiver.open();
        });

        conn.senderOpenHandler(sender -> {
            sender.setSource(sender.getRemoteSource());
            sender.open();
            futureSender.complete(sender);
        });
    }).listen(12347, res -> {
        latch.countDown();
    });
    latch.await();
}
 

@BeforeEach
public void setup() throws Exception {
    vertx = Vertx.vertx();
    server = ProtonServer.create(vertx);
    inbox = new LinkedBlockingDeque<>();
    outbox = new LinkedBlockingDeque<>();
    CountDownLatch latch = new CountDownLatch(1);
    server.connectHandler(conn -> {
        conn.closeHandler(c -> {
            conn.close();
            conn.disconnect();
        });
        conn.disconnectHandler(c -> {
            conn.disconnect();
        }).open();

        conn.sessionOpenHandler(ProtonSession::open);

        conn.receiverOpenHandler(receiver -> {
            log.debug("Receiver open");
            receiver.setTarget(receiver.getRemoteTarget());
            receiver.handler((delivery, message) -> {
                inbox.add(message);
            });
            receiver.open();
        });

        conn.senderOpenHandler(sender -> {
            vertx.setPeriodic(100, id -> {
                try {
                    Message m = outbox.poll(0, TimeUnit.SECONDS);
                    if (m != null) {
                        sender.send(m);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    // Try again later
                }
            });
            sender.open();
        });
    }).listen(0, res -> latch.countDown());
    latch.await();
    actualPort = server.actualPort();
    log.debug("Using port {}", actualPort);
}
 

@Override
public ProtonConnection sessionOpenHandler(Handler<ProtonSession> remoteSessionOpenHandler) {
  this.sessionOpenHandler = remoteSessionOpenHandler;
  return this;
}
 

@Override
public ProtonSession setIncomingCapacity(int bytes) {
  session.setIncomingCapacity(bytes);
  return this;
}
 

@Override
public ProtonSession setCondition(ErrorCondition condition) {
  session.setCondition(condition);
  return this;
}
 

/**
 * Sets a default handler on a session that is invoked when an AMQP <em>end</em> frame
 * is received from the peer.
 * <p>
 * The default handler sends an <em>end</em> frame and then frees up the resources
 * maintained for the session by invoking its <em>free</em> method.
 *
 * @param session The session to set the handler on.
 * @throws NullPointerException if session is {@code null}.
 */
public static void setDefaultCloseHandler(final ProtonSession session) {

    session.closeHandler(remoteClose -> {
        session.close();
        session.free();
    });
}
 

/**
 * Invoked when a client initiates a session (which is then opened in this method).
 * <p>
 * Subclasses should override this method if other behaviour shall be implemented on session open.
 *
 * @param con The connection of the session.
 * @param session The session that is initiated.
 */
protected void handleSessionOpen(final ProtonConnection con, final ProtonSession session) {
    log.debug("opening new session with client [container: {}]", con.getRemoteContainer());
    session.open();
}