content: /**
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.activemq.transport.http;
  
  import java.util.UUID;
  import java.util.concurrent.TimeUnit;
  import jakarta.jms.BytesMessage;
  import jakarta.jms.MapMessage;
  import jakarta.jms.MessageConsumer;
  import jakarta.jms.MessageProducer;
  import jakarta.jms.Session;
  import jakarta.jms.StreamMessage;
  import jakarta.jms.Topic;
  
  import org.apache.activemq.ActiveMQConnection;
  import org.apache.activemq.ActiveMQConnectionFactory;
  import org.apache.activemq.broker.BrokerService;
  import org.apache.activemq.broker.TransportConnector;
  import org.apache.activemq.command.ActiveMQBytesMessage;
  import org.apache.activemq.command.ActiveMQMapMessage;
  import org.apache.activemq.command.ActiveMQStreamMessage;
  import org.apache.activemq.command.ActiveMQTextMessage;
  import org.apache.activemq.util.ByteSequence;
  import org.junit.After;
  import org.junit.Before;
  import org.junit.Test;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertNotNull;
  import static org.junit.Assert.assertTrue;
  
  /**
   * This test covers the Message Compression feature of the ActiveMQConnectionFactory.setUseCompression
   * and has no relation to Http transport level compression.  The Messages are compressed using the
   * deflate algorithm by the ActiveMQ layer before marshalled to XML so only the Message body will
   * be compressed.
   */
  public class HttpSendCompressedMessagesTest {
  
      private static final Logger LOG = LoggerFactory.getLogger(HttpSendCompressedMessagesTest.class);
  
      private BrokerService broker;
      private static final String tcpBindAddress = "tcp://0.0.0.0:0";
      private static final String httpBindAddress = "http://0.0.0.0:8171";
      private ActiveMQConnectionFactory tcpConnectionFactory;
      private ActiveMQConnectionFactory httpConnectionFactory;
      private ActiveMQConnection tcpConnection;
      private ActiveMQConnection httpConnection;
      private Session tcpSession;
      private Session httpSession;
      private Topic destination;
      private MessageConsumer tcpConsumer;
      private MessageConsumer httpConsumer;
  
      private static final String destinationName = "HttpCompressionTopic";
  
      @Test
      public void testTextMessageCompressionFromTcp() throws Exception {
          sendTextMessage(true);
          doTestTextMessageCompression();
      }
  
      @Test
      public void testTextMessageCompressionFromHttp() throws Exception {
          sendTextMessage(httpConnectionFactory, true);
          doTestTextMessageCompression();
      }
  
      private void doTestTextMessageCompression() throws Exception {
          ActiveMQTextMessage tcpMessage = (ActiveMQTextMessage) tcpConsumer.receive(TimeUnit.SECONDS.toMillis(3));
          ActiveMQTextMessage httpMessage = (ActiveMQTextMessage) httpConsumer.receive(TimeUnit.SECONDS.toMillis(3));
  
          assertNotNull(tcpMessage);
          assertNotNull(httpMessage);
  
          ByteSequence tcpContent = tcpMessage.getContent();
          ByteSequence httpContent = httpMessage.getContent();
  
          assertNotNull(tcpContent);
          assertNotNull(httpContent);
  
          assertTrue(tcpMessage.isCompressed());
          assertTrue(httpMessage.isCompressed());
  
          int tcpCompressedSize = tcpContent.getLength();
          int httpCompressedSize = httpContent.getLength();
  
          assertEquals(tcpContent.getLength(), httpContent.getLength());
          assertEquals(tcpMessage.getText(), httpMessage.getText());
  
          LOG.info("Received Message on TCP: " + tcpMessage.toString());
          LOG.info("Received Message on HTTP: " + httpMessage.toString());
  
          sendTextMessage(false);
  
          ActiveMQTextMessage uncompressedHttpMessage = (ActiveMQTextMessage)
              httpConsumer.receive(TimeUnit.SECONDS.toMillis(3));
          int httpUncompressedSize = uncompressedHttpMessage.getContent().getLength();
  
          assertTrue(httpUncompressedSize > httpCompressedSize);
          assertTrue(httpUncompressedSize > tcpCompressedSize);
      }
  
      @Test
      public void testBytesMessageCompressionFromTcp() throws Exception {
          sendBytesMessage(true);
          doTestBytesMessageCompression();
      }
  
      @Test
      public void testBytesMessageCompressionFromHttp() throws Exception {
          sendBytesMessage(httpConnectionFactory, true);
          doTestBytesMessageCompression();
      }
  
      private void doTestBytesMessageCompression() throws Exception {
          ActiveMQBytesMessage tcpMessage = (ActiveMQBytesMessage) tcpConsumer.receive(TimeUnit.SECONDS.toMillis(3));
          ActiveMQBytesMessage httpMessage = (ActiveMQBytesMessage) httpConsumer.receive(TimeUnit.SECONDS.toMillis(3));
  
          assertNotNull(tcpMessage);
          assertNotNull(httpMessage);
  
          ByteSequence tcpContent = tcpMessage.getContent();
          ByteSequence httpContent = httpMessage.getContent();
  
          assertNotNull(tcpContent);
          assertNotNull(httpContent);
  
          assertTrue(tcpMessage.isCompressed());
          assertTrue(httpMessage.isCompressed());
  
          int tcpCompressedSize = tcpContent.getLength();
          int httpCompressedSize = httpContent.getLength();
  
          assertEquals(tcpContent.getLength(), httpContent.getLength());
          assertEquals(tcpMessage.readUTF(), httpMessage.readUTF());
  
          LOG.info("Received Message on TCP: " + tcpMessage.toString());
          LOG.info("Received Message on HTTP: " + httpMessage.toString());
  
          sendBytesMessage(false);
  
          ActiveMQBytesMessage uncompressedHttpMessage = (ActiveMQBytesMessage)
              httpConsumer.receive(TimeUnit.SECONDS.toMillis(3));
          int httpUncompressedSize = uncompressedHttpMessage.getContent().getLength();
  
          assertTrue(httpUncompressedSize > httpCompressedSize);
          assertTrue(httpUncompressedSize > tcpCompressedSize);
      }
  
      @Test
      public void testStreamMessageCompressionFromTcp() throws Exception {
          sendStreamMessage(true);
          doTestStreamMessageCompression();
      }
  
      @Test
      public void testStreamMessageCompressionFromHttp() throws Exception {
          sendStreamMessage(httpConnectionFactory, true);
          doTestStreamMessageCompression();
      }
  
      private void doTestStreamMessageCompression() throws Exception {
          ActiveMQStreamMessage tcpMessage = (ActiveMQStreamMessage) tcpConsumer.receive(TimeUnit.SECONDS.toMillis(3));
          ActiveMQStreamMessage httpMessage = (ActiveMQStreamMessage) httpConsumer.receive(TimeUnit.SECONDS.toMillis(3));
  
          assertNotNull(tcpMessage);
          assertNotNull(httpMessage);
  
          ByteSequence tcpContent = tcpMessage.getContent();
          ByteSequence httpContent = httpMessage.getContent();
  
          assertNotNull(tcpContent);
          assertNotNull(httpContent);
  
          assertTrue(tcpMessage.isCompressed());
          assertTrue(httpMessage.isCompressed());
  
          int tcpCompressedSize = tcpContent.getLength();
          int httpCompressedSize = httpContent.getLength();
  
          assertEquals(tcpContent.getLength(), httpContent.getLength());
          assertEquals(tcpMessage.readString(), httpMessage.readString());
  
          LOG.info("Received Message on TCP: " + tcpMessage.toString());
          LOG.info("Received Message on HTTP: " + httpMessage.toString());
  
          sendStreamMessage(false);
  
          ActiveMQStreamMessage uncompressedHttpMessage = (ActiveMQStreamMessage)
              httpConsumer.receive(TimeUnit.SECONDS.toMillis(3));
          int httpUncompressedSize = uncompressedHttpMessage.getContent().getLength();
  
          assertTrue(httpUncompressedSize > httpCompressedSize);
          assertTrue(httpUncompressedSize > tcpCompressedSize);
      }
  
      @Test
      public void testMapMessageCompressionFromTcp() throws Exception {
          sendMapMessage(true);
          doTestMapMessageCompression();
      }
  
      @Test
      public void testMapMessageCompressionFromHttp() throws Exception {
          sendMapMessage(httpConnectionFactory, true);
          doTestMapMessageCompression();
      }
  
      private void doTestMapMessageCompression() throws Exception {
          ActiveMQMapMessage tcpMessage = (ActiveMQMapMessage) tcpConsumer.receive(TimeUnit.SECONDS.toMillis(3));
          ActiveMQMapMessage httpMessage = (ActiveMQMapMessage) httpConsumer.receive(TimeUnit.SECONDS.toMillis(3));
  
          assertNotNull(tcpMessage);
          assertNotNull(httpMessage);
  
          ByteSequence tcpContent = tcpMessage.getContent();
          ByteSequence httpContent = httpMessage.getContent();
  
          assertNotNull(tcpContent);
          assertNotNull(httpContent);
  
          assertTrue(tcpMessage.isCompressed());
          assertTrue(httpMessage.isCompressed());
  
          int tcpCompressedSize = tcpContent.getLength();
          int httpCompressedSize = httpContent.getLength();
  
          assertEquals(tcpContent.getLength(), httpContent.getLength());
          assertEquals(tcpMessage.getString("content"), httpMessage.getString("content"));
  
          LOG.info("Received Message on TCP: " + tcpMessage.toString());
          LOG.info("Received Message on HTTP: " + httpMessage.toString());
  
          sendMapMessage(false);
  
          ActiveMQMapMessage uncompressedHttpMessage = (ActiveMQMapMessage)
              httpConsumer.receive(TimeUnit.SECONDS.toMillis(3));
          int httpUncompressedSize = uncompressedHttpMessage.getContent().getLength();
  
          assertTrue(httpUncompressedSize > httpCompressedSize);
          assertTrue(httpUncompressedSize > tcpCompressedSize);
      }
  
      @Before
      public void setUp() throws Exception {
          broker = new BrokerService();
          broker.setPersistent(false);
          broker.setUseJmx(false);
          broker.setAdvisorySupport(false);
          broker.setDeleteAllMessagesOnStartup(true);
          TransportConnector tcpConnector = broker.addConnector(tcpBindAddress);
          TransportConnector httpConnector = broker.addConnector(httpBindAddress);
          broker.start();
          broker.waitUntilStarted();
  
          WaitForJettyListener.waitForJettySocketToAccept(httpConnector.getPublishableConnectString());
  
          tcpConnectionFactory = new ActiveMQConnectionFactory(tcpConnector.getPublishableConnectString());
          tcpConnectionFactory.setUseCompression(true);
          httpConnectionFactory = new ActiveMQConnectionFactory(httpConnector.getPublishableConnectString());
          httpConnectionFactory.setUseCompression(true);
          tcpConnection = (ActiveMQConnection) tcpConnectionFactory.createConnection();
          httpConnection = (ActiveMQConnection) httpConnectionFactory.createConnection();
          tcpSession = tcpConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
          httpSession = httpConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
          destination = tcpSession.createTopic(destinationName);
          tcpConsumer = tcpSession.createConsumer(destination);
          httpConsumer = httpSession.createConsumer(destination);
          tcpConnection.start();
          httpConnection.start();
      }
  
      @After
      public void shutDown() throws Exception {
          tcpConnection.close();
          httpConnection.close();
  
          if (broker != null) {
              broker.stop();
              broker.waitUntilStopped();
          }
      }
  
      private void sendTextMessage(boolean compressed) throws Exception {
          sendTextMessage(tcpConnectionFactory, compressed);
      }
  
      private void sendTextMessage(ActiveMQConnectionFactory factory, boolean compressed) throws Exception {
  
          StringBuilder builder = new StringBuilder();
          for(int i = 0; i < 10; ++i) {
              builder.append(UUID.randomUUID().toString());
          }
  
          try (ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection()) {
              connection.setUseCompression(compressed);
              Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
              Topic destination = session.createTopic(destinationName);
              MessageProducer producer = session.createProducer(destination);
              producer.send(session.createTextMessage(builder.toString()));
          }
      }
  
      private void sendBytesMessage(boolean compressed) throws Exception {
          sendBytesMessage(tcpConnectionFactory, compressed);
      }
  
      private void sendBytesMessage(ActiveMQConnectionFactory factory, boolean compressed) throws Exception {
  
          StringBuilder builder = new StringBuilder();
          for(int i = 0; i < 10; ++i) {
              builder.append(UUID.randomUUID().toString());
          }
  
          try (ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection()) {
              connection.setUseCompression(compressed);
              Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
              Topic destination = session.createTopic(destinationName);
              MessageProducer producer = session.createProducer(destination);
              BytesMessage message = session.createBytesMessage();
              message.writeUTF(builder.toString());
              producer.send(message);
          }
      }
  
      private void sendStreamMessage(boolean compressed) throws Exception {
          sendStreamMessage(tcpConnectionFactory, compressed);
      }
  
      private void sendStreamMessage(ActiveMQConnectionFactory factory, boolean compressed) throws Exception {
  
          StringBuilder builder = new StringBuilder();
          for(int i = 0; i < 10; ++i) {
              builder.append(UUID.randomUUID().toString());
          }
  
          try (ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection()) {
              connection.setUseCompression(compressed);
              Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
              Topic destination = session.createTopic(destinationName);
              MessageProducer producer = session.createProducer(destination);
              StreamMessage message = session.createStreamMessage();
              message.writeString(builder.toString());
              producer.send(message);
          }
      }
  
      private void sendMapMessage(boolean compressed) throws Exception {
          sendMapMessage(tcpConnectionFactory, compressed);
      }
  
      private void sendMapMessage(ActiveMQConnectionFactory factory, boolean compressed) throws Exception {
  
          StringBuilder builder = new StringBuilder();
          for(int i = 0; i < 10; ++i) {
              builder.append(UUID.randomUUID().toString());
          }
  
          try (ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection()) {
              connection.setUseCompression(compressed);
              Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
              Topic destination = session.createTopic(destinationName);
              MessageProducer producer = session.createProducer(destination);
              MapMessage message = session.createMapMessage();
              message.setString("content", builder.toString());
              producer.send(message);
          }
      }
  }
  