VertxBasedAmqpProtocolAdapter extends AbstractProtocolAdapterBase<AmqpAdapterProperties> { protected void handleRemoteReceiverOpen(final ProtonConnection conn, final ProtonReceiver receiver) { final Device authenticatedDevice = conn.attachments().get(AmqpAdapterConstants.KEY_CLIENT_DEVICE, Device.class); final OptionalInt traceSamplingPriority = getTraceSamplingPriority(conn); final Span span = newSpan("attach receiver", authenticatedDevice, traceSamplingPriority); if (ProtonQoS.AT_MOST_ONCE.equals(receiver.getRemoteQoS())) { final Exception ex = new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "unsupported snd-settle-mode: settled"); closeLinkWithError(receiver, ex, span); } else if (receiver.getRemoteTarget() != null && receiver.getRemoteTarget().getAddress() != null) { if (!receiver.getRemoteTarget().getAddress().isEmpty()) { log.debug("closing link due to the presence of target address [{}]", receiver.getRemoteTarget().getAddress()); } final Exception ex = new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "this adapter supports anonymous relay mode only"); closeLinkWithError(receiver, ex, span); } else { receiver.setTarget(receiver.getRemoteTarget()); receiver.setSource(receiver.getRemoteSource()); receiver.setQoS(receiver.getRemoteQoS()); receiver.setPrefetch(30); receiver.setAutoAccept(false); HonoProtonHelper.setCloseHandler(receiver, remoteDetach -> onLinkDetach(receiver)); HonoProtonHelper.setDetachHandler(receiver, remoteDetach -> onLinkDetach(receiver)); receiver.handler((delivery, message) -> { try { final SpanContext spanContext = TracingHelper.extractSpanContext(tracer, message); final Span msgSpan = newSpan("upload message", authenticatedDevice, traceSamplingPriority, spanContext); msgSpan.log(Collections.singletonMap(Tags.MESSAGE_BUS_DESTINATION.getKey(), message)); final AmqpContext ctx = AmqpContext.fromMessage(delivery, message, msgSpan, authenticatedDevice); ctx.setTimer(metrics.startTimer()); final Future<Void> spanPreparationFuture = authenticatedDevice == null ? applyTraceSamplingPriorityForAddressTenant(ctx.getAddress(), msgSpan) : Future.succeededFuture(); spanPreparationFuture .compose(ar -> onMessageReceived(ctx) .onComplete(r -> msgSpan.finish())); } catch (final Exception ex) { log.warn("error handling message", ex); ProtonHelper.released(delivery, true); } }); receiver.open(); if (authenticatedDevice == null) { log.debug("established link for receiving messages from device [container: {}]", conn.getRemoteContainer()); } else { log.debug("established link for receiving messages from device [tenant: {}, device-id: {}]]", authenticatedDevice.getTenantId(), authenticatedDevice.getDeviceId()); } span.log("link established"); } span.finish(); }  @Autowired void setMetrics(final AmqpAdapterMetrics metrics); @Override int getPortDefaultValue(); @Override int getInsecurePortDefaultValue();  }
@Test public void testAdapterAcceptsAnonymousRelayReceiverOnly() { final ProtonServer server = getAmqpServer(); final VertxBasedAmqpProtocolAdapter adapter = getAdapter(server); final ResourceIdentifier targetAddress = ResourceIdentifier.from(TelemetryConstants.TELEMETRY_ENDPOINT, TEST_TENANT_ID, TEST_DEVICE); final ProtonReceiver link = getReceiver(ProtonQoS.AT_LEAST_ONCE, getTarget(targetAddress)); adapter.handleRemoteReceiverOpen(getConnection(null), link); verify(link).close(); }