private static Future<ResourceIdentifier> getResourceIdentifier(final Source source) {

        if (source == null) {
            return Future.failedFuture(new ClientErrorException(HttpURLConnection.HTTP_NOT_FOUND, "no such node"));
        } else {
            final Promise<ResourceIdentifier> result = Promise.promise();
            try {
                if (Strings.isNullOrEmpty(source.getAddress())) {
                    result.fail(new ClientErrorException(HttpURLConnection.HTTP_NOT_FOUND,
                            "no such node"));
                } else {
                    result.complete(ResourceIdentifier.fromString(source.getAddress()));
                }
            } catch (Throwable e) {
                result.fail(e);
            }
            return result.future();
        }
    }
 

/**
 * Handles a request from a client to establish a link for receiving messages from this server.
 *
 * @param con the connection to the client.
 * @param sender the sender created for the link.
 */
@Override
protected void handleSenderOpen(final ProtonConnection con, final ProtonSender sender) {

    final Source remoteSource = sender.getRemoteSource();
    LOG.debug("client [{}] wants to open a link for receiving messages [address: {}]",
            con.getRemoteContainer(), remoteSource);
    try {
        final ResourceIdentifier targetResource = getResourceIdentifier(remoteSource.getAddress());
        final AmqpEndpoint endpoint = getEndpoint(targetResource);

        if (endpoint == null) {
            LOG.debug("no endpoint registered for node [{}]", targetResource);
            con.setCondition(ProtonHelper.condition(AmqpError.NOT_FOUND, "no such node")).close();
        } else {
            final HonoUser user = Constants.getClientPrincipal(con);
            if (Constants.SUBJECT_ANONYMOUS.equals(user.getName())) {
                con.setCondition(ProtonHelper.condition(AmqpError.UNAUTHORIZED_ACCESS, "client must authenticate using SASL")).close();
            } else {
                Constants.copyProperties(con, sender);
                sender.setSource(sender.getRemoteSource());
                endpoint.onLinkAttach(con, sender, targetResource);
            }
        }
    } catch (final IllegalArgumentException e) {
        LOG.debug("client has provided invalid resource identifier as source address", e);
        con.setCondition(ProtonHelper.condition(AmqpError.INVALID_FIELD, "malformed source address")).close();
    }
}
 

/**
 * Verifies that the attempt to create a receiver fails with a
 * {@code ServerErrorException} if the connection gets disconnected
 * before the remote peer has sent its attach frame. It is verified
 * that this is done before the link establishment timeout.
 *
 * @param ctx The vert.x test context.
 */
@Test
public void testCreateReceiverFailsOnDisconnectBeforeOpen(final VertxTestContext ctx) {

    final long linkEstablishmentTimeout = 444L; // choose a distinct value here
    props.setLinkEstablishmentTimeout(linkEstablishmentTimeout);
    // don't run linkEstablishmentTimeout timer handler
    when(vertx.setTimer(eq(linkEstablishmentTimeout), VertxMockSupport.anyHandler())).thenAnswer(invocation -> 0L);

    final Source source = mock(Source.class);
    when(source.getAddress()).thenReturn("source/address");
    final ProtonReceiver receiver = mock(ProtonReceiver.class);
    when(receiver.isOpen()).thenReturn(Boolean.TRUE);
    when(receiver.getSource()).thenReturn(source);
    when(receiver.getRemoteSource()).thenReturn(source);
    when(con.createReceiver(anyString())).thenReturn(receiver);

    final Handler<String> remoteCloseHook = VertxMockSupport.mockHandler();

    // GIVEN an established connection
    honoConnection.connect()
        .compose(c -> {
            // WHEN creating a receiver link with a close hook
            final Future<ProtonReceiver> result = honoConnection.createReceiver("source", ProtonQoS.AT_LEAST_ONCE,
                    mock(ProtonMessageHandler.class), remoteCloseHook);
            // THEN the result is not completed at first
            ctx.verify(() -> assertThat(result.isComplete()).isFalse());
            // WHEN the downstream connection fails
            connectionFactory.getDisconnectHandler().handle(con);
            return result;
        })
        // THEN the attempt is failed
        .onComplete(ctx.failing(t -> {
            ctx.verify(() -> assertThat(((ServerErrorException) t).getErrorCode()).isEqualTo(HttpURLConnection.HTTP_UNAVAILABLE));
            ctx.completeNow();
        }));
}
 

/**
 * Handles a request from a client to establish a link for receiving messages from this server.
 *
 * @param con the connection to the client.
 * @param sender the sender created for the link.
 */
protected void handleSenderOpen(final ProtonConnection con, final ProtonSender sender) {
    final Source remoteSource = sender.getRemoteSource();
    log.debug("client [container: {}] wants to open a link [address: {}] for receiving messages",
            con.getRemoteContainer(), remoteSource);
    try {
        final ResourceIdentifier targetResource = getResourceIdentifier(remoteSource.getAddress());
        final AmqpEndpoint endpoint = getEndpoint(targetResource);
        if (endpoint == null) {
            handleUnknownEndpoint(con, sender, targetResource);
        } else {
            final HonoUser user = Constants.getClientPrincipal(con);
            getAuthorizationService().isAuthorized(user, targetResource, Activity.READ).onComplete(authAttempt -> {
                if (authAttempt.succeeded() && authAttempt.result()) {
                    Constants.copyProperties(con, sender);
                    sender.setSource(sender.getRemoteSource());
                    sender.setTarget(sender.getRemoteTarget());
                    endpoint.onLinkAttach(con, sender, targetResource);
                } else {
                    log.debug("subject [{}] is not authorized to READ from [{}]", user.getName(), targetResource);
                    sender.setCondition(ProtonHelper.condition(AmqpError.UNAUTHORIZED_ACCESS.toString(), "unauthorized"));
                    sender.close();
                }
            });
        }
    } catch (final IllegalArgumentException e) {
        log.debug("client has provided invalid resource identifier as target address", e);
        sender.setCondition(ProtonHelper.condition(AmqpError.NOT_FOUND, "no such address"));
        sender.close();
    }
}
 

private String getAddress(Source source) {
    if (source == null) {
        return null;
    } else {
        return source.getAddress();
    }
}
 

private void testHandlerCallsCloseHook(
        final VertxTestContext ctx,
        final BiConsumer<ProtonReceiver, ArgumentCaptor<Handler<AsyncResult<ProtonReceiver>>>> handlerCaptor) {

    // GIVEN an established connection
    final Source source = mock(Source.class);
    when(source.getAddress()).thenReturn("source/address");
    final ProtonReceiver receiver = mock(ProtonReceiver.class);
    when(receiver.isOpen()).thenReturn(Boolean.TRUE);
    when(receiver.getSource()).thenReturn(source);
    when(receiver.getRemoteSource()).thenReturn(source);
    when(con.createReceiver(anyString())).thenReturn(receiver);

    final Handler<String> remoteCloseHook = VertxMockSupport.mockHandler();
    final ArgumentCaptor<Handler<AsyncResult<ProtonReceiver>>> captor = VertxMockSupport.argumentCaptorHandler();

    honoConnection.connect()
        .compose(c -> {

            // WHEN creating a receiver link with a close hook

            final Future<ProtonReceiver> r = c.createReceiver(
                    "source",
                    ProtonQoS.AT_LEAST_ONCE,
                    mock(ProtonMessageHandler.class),
                    remoteCloseHook);

            // wait for peer's attach frame
            final ArgumentCaptor<Handler<AsyncResult<ProtonReceiver>>> openHandlerCaptor = VertxMockSupport.argumentCaptorHandler();
            ctx.verify(() -> verify(receiver).openHandler(openHandlerCaptor.capture()));
            openHandlerCaptor.getValue().handle(Future.succeededFuture(receiver));

            return r;
        })
        .onComplete(ctx.succeeding(recv -> {

            // WHEN the peer sends a detach frame
            handlerCaptor.accept(receiver, captor);
            captor.getValue().handle(Future.succeededFuture(receiver));

            ctx.verify(() -> {
                // THEN the close hook is called
                verify(remoteCloseHook).handle(any());

                // and the receiver link is closed
                verify(receiver).close();
                verify(receiver).free();
            });
            ctx.completeNow();
        }));

}
 

private ProtonSender getSender(final String sourceAddress) {
    final ProtonSender sender = mock(ProtonSender.class);
    when(sender.getRemoteSource()).thenReturn(mock(Source.class));
    when(sender.getRemoteSource().getAddress()).thenReturn(sourceAddress);
    return sender;
}
 

@Override
public Source getRemoteSource()
{
    return _remoteSource;
}
 

void setRemoteSource(Source source)
{
    _remoteSource = source;
}
 

@Override
public Source getSource()
{
    return _source;
}
 

@Override
public void setSource(Source source)
{
    // TODO - should be an error if local state is ACTIVE
    _source = source;
}
 

public Attach newInstance(Object described)
{
    List l = (List) described;

    Attach o = new Attach();

    if(l.size() <= 2)
    {
        throw new DecodeException("The role field cannot be omitted");
    }

    switch(14 - l.size())
    {

        case 0:
            o.setProperties( (Map) l.get( 13 ) );
        case 1:
            Object val1 = l.get( 12 );
            if( val1 == null || val1.getClass().isArray() )
            {
                o.setDesiredCapabilities( (Symbol[]) val1 );
            }
            else
            {
                o.setDesiredCapabilities( (Symbol) val1 );
            }
        case 2:
            Object val2 = l.get( 11 );
            if( val2 == null || val2.getClass().isArray() )
            {
                o.setOfferedCapabilities( (Symbol[]) val2 );
            }
            else
            {
                o.setOfferedCapabilities( (Symbol) val2 );
            }
        case 3:
            o.setMaxMessageSize( (UnsignedLong) l.get( 10 ) );
        case 4:
            o.setInitialDeliveryCount( (UnsignedInteger) l.get( 9 ) );
        case 5:
            Boolean incompleteUnsettled = (Boolean) l.get(8);
            o.setIncompleteUnsettled(incompleteUnsettled == null ? false : incompleteUnsettled);
        case 6:
            o.setUnsettled( (Map) l.get( 7 ) );
        case 7:
            o.setTarget( (Target) l.get( 6 ) );
        case 8:
            o.setSource( (Source) l.get( 5 ) );
        case 9:
            UnsignedByte rcvSettleMode = (UnsignedByte) l.get(4);
            o.setRcvSettleMode(rcvSettleMode == null ? ReceiverSettleMode.FIRST : ReceiverSettleMode.values()[rcvSettleMode.intValue()]);
        case 10:
            UnsignedByte sndSettleMode = (UnsignedByte) l.get(3);
            o.setSndSettleMode(sndSettleMode == null ? SenderSettleMode.MIXED : SenderSettleMode.values()[sndSettleMode.intValue()]);
        case 11:
            o.setRole( Boolean.TRUE.equals( l.get( 2 ) ) ? Role.RECEIVER : Role.SENDER);
        case 12:
            o.setHandle( (UnsignedInteger) l.get( 1 ) );
        case 13:
            o.setName( (String) l.get( 0 ) );
    }


    return o;
}
 

@Override
public ProtonPublisher<Message> setSource(Source source) {
  delegate.setSource(source);
  return this;
}
 

@Override
public Source getSource() {
  return delegate.getSource();
}
 

@Override
public Source getRemoteSource() {
  return delegate.getRemoteSource();
}
 

@Override
public ProtonSubscriber<Tracker> setSource(Source source) {
  sender.setSource(source);
  return this;
}
 

@Override
public Source getSource() {
  return sender.getSource();
}
 

public Source getRemoteSource() {
  return sender.getRemoteSource();
}
 

public ProtonPublisherImpl(String address, ProtonConnectionImpl conn, ProtonPublisherOptions options) {
  this.connCtx = conn.getContext();
  this.conn = conn;

  ProtonLinkOptions linkOptions = new ProtonLinkOptions();
  if(options.getLinkName() != null) {
    linkOptions.setLinkName(options.getLinkName());
  }

  receiver = conn.createReceiver(address, linkOptions);
  receiver.setAutoAccept(false);
  receiver.setPrefetch(0);

  if(options.getMaxOutstandingCredit() > 0) {
    maxOutstandingCredit = options.getMaxOutstandingCredit();
  }

  org.apache.qpid.proton.amqp.messaging.Source source = (org.apache.qpid.proton.amqp.messaging.Source) receiver.getSource();
  durable = options.isDurable();
  if(durable) {
    source.setExpiryPolicy(TerminusExpiryPolicy.NEVER);
    source.setDurable(TerminusDurability.UNSETTLED_STATE);
  }

  if(options.isDynamic()) {
    source.setAddress(null);
    source.setDynamic(true);
  }

  ArrayList<Symbol> capabilities = new ArrayList<>();
  if(options.isShared()) {
    capabilities.add(SHARED);
  }
  if(options.isGlobal()) {
    capabilities.add(GLOBAL);
  }

  if(!capabilities.isEmpty()) {
    Symbol[] caps = capabilities.toArray(new Symbol[capabilities.size()]);
    source.setCapabilities(caps);
  }
}
 

@Override
public ProtonPublisher<Delivery> setSource(Source source) {
  receiver.setSource(source);
  return this;
}
 

@Override
public Source getSource() {
  return receiver.getSource();
}
 

@Override
public Source getRemoteSource() {
  return receiver.getRemoteSource();
}
 

@Override
public String getRemoteAddress() {
  Source remoteSource = getRemoteSource();

  return remoteSource == null ? null : remoteSource.getAddress();
}
 

@Override
public ProtonSubscriber<Message> setSource(Source source) {
  delegate.setSource(source);
  return this;
}
 

@Override
public Source getSource() {
  return delegate.getSource();
}
 

public Source getRemoteSource() {
  return delegate.getRemoteSource();
}
 

@Override
public Source getRemoteSource() {
  return link.getRemoteSource();
}
 

@Override
public Source getSource() {
  return link.getSource();
}
 

@Override
public T setSource(Source source) {
  link.setSource(source);
  return self();
}
 

@Override
public String getRemoteAddress() {
  Source remoteSource = getRemoteSource();

  return remoteSource == null ? null : remoteSource.getAddress();
}