MessageHelper { public static String getPayloadAsString(final Message message) { Objects.requireNonNull(message); if (message.getBody() == null) { LOG.trace("message has no body"); return null; } if (message.getBody() instanceof Data) { final Data body = (Data) message.getBody(); return Buffer.buffer(body.getValue().getArray()).toString(); } else if (message.getBody() instanceof AmqpValue) { final AmqpValue body = (AmqpValue) message.getBody(); if (body.getValue() instanceof byte[]) { return Buffer.buffer((byte[]) body.getValue()).toString(); } else if (body.getValue() instanceof String) { return (String) body.getValue(); } } LOG.debug("unsupported body type [{}]", message.getBody().getClass().getName()); return null; } private  MessageHelper(); static String getDeviceId(final Message msg); static String getDeviceId(final ApplicationProperties properties); static String getTenantId(final Message msg); static String getTenantId(final ApplicationProperties properties); static String getGatewayId(final Message msg); static String getGatewayId(final ApplicationProperties properties); static String getRegistrationAssertion(final Message msg); static String getAndRemoveRegistrationAssertion(final Message msg); static String getDeviceIdAnnotation(final Message msg); static String getTenantIdAnnotation(final Message msg); @SuppressWarnings("unchecked") static T getApplicationProperty(final ApplicationProperties props, final String name,
            final Class<T> type); static JsonObject getJsonPayload(final Message msg); static Buffer getPayload(final Message msg); static String getPayloadAsString(final Message message); static void addTenantId(final Message msg, final String tenantId); static void addDeviceId(final Message msg, final String deviceId); static void addRegistrationAssertion(final Message msg, final String token); static void addCacheDirective(final Message msg, final CacheDirective cacheDirective); static String getCacheDirective(final Message msg); static void setTimeToLive(final Message message, final Duration timeToLive); static void addTimeUntilDisconnect(final Message msg, final int timeUntilDisconnect); static Integer getTimeUntilDisconnect(final Message msg); static Integer getStatus(final Message msg); static void addStatus(final Message msg, final int status); static void addProperty(final Message msg, final String key, final Object value); static void rejected(final ProtonDelivery delivery, final ErrorCondition error); static void annotate(final Message msg, final ResourceIdentifier resourceIdentifier); static void addJmsVendorProperties(final Message msg); static void addAnnotation(final Message msg, final String key, final Object value); @SuppressWarnings("unchecked") static T getAnnotation(final Message msg, final String key, final Class<T> type); static void setCreationTime(final Message msg); static boolean isDeviceCurrentlyConnected(final Message msg); static void setJsonPayload(final Message message, final JsonObject payload); static void setJsonPayload(final Message message, final String payload); static void setPayload(final Message message, final String contentType, final Buffer payload); static void setPayload(final Message message, final String contentType, final byte[] payload); static boolean hasDataBody(final Message message); static boolean hasAmqpValueBody(final Message message); static Message getShallowCopy(final Message message); static Object getCorrelationId(final Message message); static Message newMessage(
            final QoS qos,
            final ResourceIdentifier target,
            final String contentType,
            final Buffer payload,
            final TenantObject tenant,
            final Duration timeToLive,
            final String adapterTypeName); static Message newMessage(
            final QoS qos,
            final ResourceIdentifier target,
            final String publishAddress,
            final String contentType,
            final Buffer payload,
            final TenantObject tenant,
            final JsonObject deviceDefaultProperties,
            final Integer timeUntilDisconnect,
            final Duration timeToLive,
            final String adapterTypeName,
            final boolean addDefaults,
            final boolean addJmsVendorProps); static Message addProperties(
            final Message msg,
            final QoS qos,
            final ResourceIdentifier target,
            final String publishAddress,
            final TenantObject tenant,
            final JsonObject deviceDefaultProperties,
            final Integer timeUntilDisconnect,
            final Duration timeToLive,
            final String adapterTypeName,
            final boolean addDefaults,
            final boolean addJmsVendorProps); static Message addDefaults(
            final Message message,
            final ResourceIdentifier target,
            final JsonObject defaults,
            final long maxTtl); static final String ANNOTATION_X_OPT_RETAIN; static final String APP_PROPERTY_CACHE_CONTROL; static final String APP_PROPERTY_DEVICE_ID; static final String APP_PROPERTY_DEVICE_TTD; static final String APP_PROPERTY_GATEWAY_ID; static final String APP_PROPERTY_ADAPTER_INSTANCE_ID; static final String APP_PROPERTY_LIFESPAN; static final String APP_PROPERTY_ORIG_ADDRESS; static final String APP_PROPERTY_ORIG_ADAPTER; static final String APP_PROPERTY_QOS; static final String APP_PROPERTY_REGISTRATION_ASSERTION; static final String APP_PROPERTY_RESOURCE; static final String APP_PROPERTY_STATUS; static final String APP_PROPERTY_TENANT_ID; static final String APP_PROPERTY_RESOURCE_VERSION; static final String APP_PROPERTY_CMD_VIA; static final String SYS_HEADER_PROPERTY_DELIVERY_COUNT; static final String SYS_HEADER_PROPERTY_DURABLE; static final String SYS_HEADER_PROPERTY_FIRST_ACQUIRER; static final String SYS_HEADER_PROPERTY_PRIORITY; static final String SYS_HEADER_PROPERTY_TTL; static final String SYS_PROPERTY_ABSOLUTE_EXPIRY_TIME; static final String SYS_PROPERTY_CONTENT_ENCODING; static final String SYS_PROPERTY_CONTENT_TYPE; static final String SYS_PROPERTY_CORRELATION_ID; static final String SYS_PROPERTY_CREATION_TIME; static final String SYS_PROPERTY_GROUP_ID; static final String SYS_PROPERTY_GROUP_SEQUENCE; static final String SYS_PROPERTY_MESSAGE_ID; static final String SYS_PROPERTY_REPLY_TO; static final String SYS_PROPERTY_REPLY_TO_GROUP_ID; static final String SYS_PROPERTY_SUBJECT; static final String SYS_PROPERTY_USER_ID; static final String SYS_PROPERTY_TO; static final int TTD_VALUE_UNLIMITED; static final String JMS_VENDOR_PROPERTY_CONTENT_ENCODING; static final String JMS_VENDOR_PROPERTY_CONTENT_TYPE; static final String CONTENT_TYPE_APPLICATION_JSON; static final String CONTENT_TYPE_OCTET_STREAM; static final String CONTENT_TYPE_TEXT_PLAIN; }
@Test public void testGetPayloadAsStringHandlesNonCharacterPayload() { final Message msg = ProtonHelper.message(); msg.setBody(new Data(new Binary(new byte[] { (byte) 0xc3, (byte) 0x28 }))); assertThat(MessageHelper.getPayloadAsString(msg)).isNotNull(); msg.setBody(new Data(new Binary(new byte[] { (byte) 0xf0, (byte) 0x28, (byte) 0x8c, (byte) 0xbc }))); assertThat(MessageHelper.getPayloadAsString(msg)).isNotNull(); }