### Question:
RabbitMQMessage implements Message { @Override public void readMessage(final MessageReader messageReader) { if (blobstoreApi != null) { BlobStoreLink blobStoreLink = MessagingSerializationHelper.readObject( serializationManager, BlobStoreLink.class, delivery.getBody()); if (blobstoreApi.isExists(blobStoreLink.getNamespace(), blobStoreLink.getKey())) { blobstoreApi.readBlob(blobStoreLink.getNamespace(), blobStoreLink.getKey(), in -> { if (gzip) { try (GZIPInputStream gzipInputStream = new GZIPInputStream(in)) { messageReader.read(gzipInputStream); } catch (IOException e) { throw new IllegalStateException(e); } } else { messageReader.read(in); } }); } else { throw new IllegalStateException( "Failed reading message with id (id doesn't exist)" + blobStoreLink.getNamespace() + ':' + blobStoreLink.getKey()); } } else { byte[] body = delivery.getBody(); try (InputStream stream = getInputStream(body)) { messageReader.read(stream); } catch (IOException e) { throw new IllegalStateException(e); } } } RabbitMQMessage(
            SerializationManager serializationManager, QueueingConsumer.Delivery delivery,
            BlobStoreAPI blobstoreApi, boolean gzip,
            boolean logMessageWhileReading); @Override String getMessageHeader(String headerName); void addErrorMessageHeader(String microserviceURI, long timestamp, int errorCode, String errorMessage); void setFailed(boolean failed); boolean isFailed(); @Override Map<String, String> getMessageHeaders(); @Override Map<String, String> getMessageContext(); @Override String getContextValue(String key); @Override void readMessage(final MessageReader messageReader); @Override Object getUnderlyingMessageObject(); @Override T readObject(Class<T> format); List<XDeath> getXDeaths(); @NoJavadoc static String getNamespace(String blobstoreNamespace, String exchange, String queue); @NoJavadoc AMQP.BasicProperties getProperties();  }### Answer:
@Test public void testThatWeGetGZipStreamForCompressedData() throws Exception { try (ByteArrayOutputStream s = new ByteArrayOutputStream()) { try (GZIPOutputStream out = new GZIPOutputStream(s)) { out.write(data); out.finish(); QueueingConsumer.Delivery delivery = new QueueingConsumer.Delivery(null, null, s.toByteArray()); RabbitMQMessage mqMsg = new RabbitMQMessage(new SerializationManagerImpl(), delivery, null, true, false); mqMsg.readMessage(in -> assertTrue(in instanceof GZIPInputStream)); } } }
@Test public void testThatWeGetNormalStreamForUncompressedData() throws Exception { QueueingConsumer.Delivery delivery = new QueueingConsumer.Delivery(null, null, data); RabbitMQMessage mqMsg = new RabbitMQMessage(new SerializationManagerImpl(), delivery, null, false, false); mqMsg.readMessage(in -> assertFalse(in instanceof GZIPInputStream)); }
@Test public void testThatWeGetLoggingStreamForUncompressedData() throws Exception { QueueingConsumer.Delivery delivery = new QueueingConsumer.Delivery(null, null, data); RabbitMQMessage mqMsg = new RabbitMQMessage(new SerializationManagerImpl(), delivery, null, false, true); mqMsg.readMessage(in -> assertTrue(in instanceof LoggingInputStream)); }
@Test public void testThatWeGetLoggingStreamForCompressedData() throws Exception { try (ByteArrayOutputStream s = new ByteArrayOutputStream()) { try (GZIPOutputStream out = new GZIPOutputStream(s)) { out.write(data); out.finish(); QueueingConsumer.Delivery delivery = new QueueingConsumer.Delivery(null, null, s.toByteArray()); RabbitMQMessage mqMsg = new RabbitMQMessage(new SerializationManagerImpl(), delivery, null, true, true); mqMsg.readMessage(in -> assertTrue(in instanceof LoggingInputStream)); } } }