SoakPagingTest.java

/*
 * 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.artemis.tests.smoke.replicationflow;

import javax.jms.BytesMessage;
import javax.jms.CompletionListener;
import javax.jms.Connection;
import javax.jms.ConnectionConsumer;
import javax.jms.ConnectionFactory;
import javax.jms.ConnectionMetaData;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSContext;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
import javax.jms.ServerSessionPool;
import javax.jms.Session;
import javax.jms.StreamMessage;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;
import java.io.Serializable;
import java.util.Arrays;
import java.util.Collection;
import java.util.Enumeration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.activemq.artemis.tests.smoke.common.SmokeTestBase;
import org.apache.activemq.artemis.utils.ExecuteUtil;
import org.apache.activemq.artemis.utils.SpawnedVMSupport;
import org.apache.qpid.jms.JmsConnectionFactory;
import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;
import org.jboss.logging.Logger;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

@RunWith(Parameterized.class)
public class SoakPagingTest extends SmokeTestBase {

   private static final Logger log = Logger.getLogger(SoakPagingTest.class);

   public static final int LAG_CONSUMER_TIME = 1000;
   public static final int TIME_RUNNING = 4000;
   public static final int CLIENT_KILLS = 2;

   String protocol;
   String consumerType;
   boolean transaction;
   final String destination;

   public SoakPagingTest(String protocol, String consumerType, boolean transaction) {
      this.protocol = protocol;
      this.consumerType = consumerType;
      this.transaction = transaction;

      if (consumerType.equals("queue")) {
         destination = "exampleQueue";
      } else {
         destination = "exampleTopic";
      }
   }

   @Parameterized.Parameters(name = "protocol={0}, type={1}, tx={2}")
   public static Collection<Object[]> getParams() {
      return Arrays.asList(new Object[][]{{"MQTT", "topic", false}, {"AMQP", "shared", false}, {"AMQP", "queue", false}, {"OPENWIRE", "topic", false}, {"OPENWIRE", "queue", false}, {"CORE", "shared", false}, {"CORE", "queue", false},
         {"AMQP", "shared", true}, {"AMQP", "queue", true}, {"OPENWIRE", "topic", true}, {"OPENWIRE", "queue", true}, {"CORE", "shared", true}, {"CORE", "queue", true}});
   }

   public static final String SERVER_NAME_0 = "replicated-static0";
   public static final String SERVER_NAME_1 = "replicated-static1";

   static AtomicInteger produced = new AtomicInteger(0);
   static AtomicInteger consumed = new AtomicInteger(0);
   private static Process server0;

   private static Process server1;
   @Before
   public void before() throws Exception {
      cleanupData(SERVER_NAME_0);
      cleanupData(SERVER_NAME_1);

      server0 = startServer(SERVER_NAME_0, 0, 30000);
   }

   static final int consumer_threads = 20;
   static final int producer_threads = 20;
   static AtomicInteger producer_count = new AtomicInteger(0);
   static AtomicInteger consumer_count = new AtomicInteger(0);

   private static ConnectionFactory createConnectionFactory(String protocol, String uri) {
      if (protocol.toUpperCase().equals("OPENWIRE")) {
         return new org.apache.activemq.ActiveMQConnectionFactory("failover:(" + uri + ")");
      } else if (protocol.toUpperCase().equals("MQTT")) {
         return new MQTTCF();
      } else if (protocol.toUpperCase().equals("AMQP")) {
         return new JmsConnectionFactory("failover:(amqp://localhost:61616,amqp://localhost:61617)?failover.maxReconnectAttempts=16&jms.prefetchPolicy.all=5&jms.forceSyncSend=true");
      } else if (protocol.toUpperCase().equals("CORE") || protocol.toUpperCase().equals("ARTEMIS")) {
         return new org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory(uri);
      } else {
         throw new IllegalStateException("Unknown:" + protocol);
      }
   }

   public static void main(String[] arg) {
      try {

         if (arg.length != 4) {
            System.err.println("You need to pass in protocol, consumerType, Time, transaction");
            System.exit(0);
         }

         String protocol = arg[0];
         String consumerType = arg[1];
         int time = Integer.parseInt(arg[2]);
         boolean tx = Boolean.parseBoolean(arg[3]);
         if (time == 0) {
            time = 15000;
         }

         final String host = "localhost";
         final int port = 61616;

         final ConnectionFactory factory = createConnectionFactory(protocol, "tcp://" + host + ":" + port);

         CountDownLatch producersLatch = new CountDownLatch(producer_threads);
         CountDownLatch consumersLatch = new CountDownLatch(consumer_threads);

         for (int i = 0; i < producer_threads; i++) {
            Thread t = new Thread(new Runnable() {
               @Override
               public void run() {
                  SoakPagingTest app = new SoakPagingTest(protocol, consumerType, tx);
                  app.produce(factory, producer_count.incrementAndGet(), producersLatch);
               }
            });
            t.start();
         }

         Thread.sleep(1000);

         for (int i = 0; i < consumer_threads; i++) {
            Thread t = new Thread(new Runnable() {
               @Override
               public void run() {
                  SoakPagingTest app = new SoakPagingTest(protocol, consumerType, tx);
                  app.consume(factory, consumer_count.getAndIncrement(), consumersLatch);
               }
            });
            t.start();
         }

         log.debug("Awaiting producers...");
         if (!producersLatch.await(30000, TimeUnit.MILLISECONDS)) {
            System.err.println("Awaiting producers timeout");
            System.exit(0);
         }

         log.debug("Awaiting consumers...");
         if (!consumersLatch.await(30000, TimeUnit.MILLISECONDS)) {
            System.err.println("Awaiting consumers timeout");
            System.exit(0);
         }

         log.debug("Awaiting timeout...");
         Thread.sleep(time);

         int exitStatus = consumed.get() > 0 ? 1 : 0;
         log.debug("Exiting with the status: " + exitStatus);
         System.exit(exitStatus);
      } catch (Throwable t) {
         System.err.println("Exiting with the status 0. Reason: " + t);
         t.printStackTrace();
         System.exit(0);
      }

   }

   @Test
   public void testPagingReplication() throws Throwable {

      server1 = startServer(SERVER_NAME_1, 0, 30000);

      for (int i = 0; i < CLIENT_KILLS; i++) {
         Process process = SpawnedVMSupport.spawnVM(SoakPagingTest.class.getName(), protocol, consumerType, "" + TIME_RUNNING, "" + transaction);

         int result = process.waitFor();
         if (result <= 0) {
            jstack();
         }
         Assert.assertTrue(result > 0);
      }
   }

   protected void jstack() throws Exception {
      try {
         System.out.println("*******************************************************************************************************************************");
         System.out.println("SERVER 0 jstack");
         System.out.println("*******************************************************************************************************************************");
         ExecuteUtil.runCommand(true, 1, TimeUnit.MINUTES, "jstack", "" + ExecuteUtil.getPID(server0));
      } catch (Throwable e) {
         log.warn("Error executing jstack on Server 0", e);
      }
      try {
         System.out.println("*******************************************************************************************************************************");
         System.out.println("SERVER 1 jstack");
         System.out.println("*******************************************************************************************************************************");
         ExecuteUtil.runCommand(true, 1, TimeUnit.MINUTES, "jstack", "" + ExecuteUtil.getPID(server1));
      } catch (Throwable e) {
         log.warn("Error executing jstack on Server 1", e);
      }
   }

   public void produce(ConnectionFactory factory, int index, CountDownLatch latch) {
      try {

         StringBuffer bufferlarge = new StringBuffer();
         while (bufferlarge.length() < 110000) {
            bufferlarge.append("asdflkajdhsf akljsdfh akljsdfh alksjdfh alkdjsf ");
         }
         Connection connection = factory.createConnection("admin", "admin");

         latch.countDown();

         connection.start();
         log.debug("Producer" + index + " started");

         final Session session;

         if (transaction) {
            session = connection.createSession(true, Session.SESSION_TRANSACTED);
         } else {
            session = connection.createSession(false, Session.DUPS_OK_ACKNOWLEDGE);
         }

         Destination address;

         if (consumerType.equals("queue")) {
            address = session.createQueue(destination);
         } else {
            address = session.createTopic(destination);
         }

         MessageProducer messageProducer = session.createProducer(address);

         int i = 0;
         while (true) {

            Message message;
            if (i % 100 == 0) {
               message = session.createTextMessage(bufferlarge.toString());
            } else {
               message = session.createTextMessage("fkjdslkfjdskljf;lkdsjf;kdsajf;lkjdf;kdsajf;kjdsa;flkjdsa;lfkjdsa;flkj;dsakjf;dsajf;askjd;fkj;dsajflaskfja;fdlkajs;lfdkja;kfj;dsakfj;akdsjf;dsakjf;akfj;lakdsjf;lkasjdf;ksajf;kjdsa;fkj;adskjf;akdsjf;kja;sdkfj;akdsjf;akjdsf;adskjf;akdsjf;askfj;aksjfkdjafndmnfmdsnfjadshfjdsalkfjads;fkjdsa;kfja;skfj;akjfd;akjfd;ksaj;fkja;kfj;dsakjf;dsakjf;dksjf;akdsjf;kdsajf");
            }

            messageProducer.send(message);
            produced.incrementAndGet();
            i++;
            if (i % 100 == 0) {
               log.debug("Producer" + index + " published " + i + " messages");
               if (transaction) {
                  session.commit();
               }
            }
         }
      } catch (Exception e) {
         System.err.println("Error on Producer" + index + ": " + e.getMessage());
         e.printStackTrace();
      }
   }

   public void consume(ConnectionFactory factory, int index, CountDownLatch latch) {
      try {
         Connection connection = factory.createConnection("admin", "admin");

         final Session session;

         if (transaction) {
            session = connection.createSession(true, Session.SESSION_TRANSACTED);
         } else {
            session = connection.createSession(false, Session.DUPS_OK_ACKNOWLEDGE);
         }

         Destination address;

         if (consumerType.equals("queue")) {
            address = session.createQueue(destination);
         } else {
            address = session.createTopic(destination);
         }

         String consumerId = "ss" + (index % 5);
         MessageConsumer messageConsumer;

         if (protocol.equals("shared")) {
            messageConsumer = session.createSharedConsumer((Topic)address, consumerId);
         } else {
            messageConsumer = session.createConsumer(address);
         }

         if (LAG_CONSUMER_TIME > 0) Thread.sleep(LAG_CONSUMER_TIME);

         latch.countDown();
         connection.start();
         log.debug("Consumer" + index + " started");

         int i = 0;
         while (true) {
            Message m = messageConsumer.receive(1000);
            consumed.incrementAndGet();
            if (m == null)
               log.debug("Consumer" + index + "received null");
            i++;
            if (i % 100 == 0) {
               log.debug("Consumer" + index + "received " + i + " messages");
               if (transaction) {
                  session.commit();
               }
            }
         }
      } catch (Exception e) {
         System.err.println("Error on Consumer" + index + ": " + e.getMessage());
         e.printStackTrace();
      }
   }
}

class MQTTCF implements ConnectionFactory, Connection, Session, Topic, MessageConsumer, MessageProducer {

   final MQTT mqtt = new MQTT();
   private String topicName;
   private BlockingConnection blockingConnection;
   private boolean consumer;

   MQTTCF() {
      try {
         mqtt.setHost("localhost", 61616);
      } catch (Exception ignored) {
      }
   }

   @Override
   public Connection createConnection() throws JMSException {
      return new MQTTCF();
   }

   @Override
   public Connection createConnection(String userName, String password) throws JMSException {
      MQTTCF result = new MQTTCF();
      result.mqtt.setUserName(userName);
      result.mqtt.setPassword(password);
      return result;
   }

   @Override
   public JMSContext createContext() {
      return null;
   }

   @Override
   public JMSContext createContext(int sessionMode) {
      return null;
   }

   @Override
   public JMSContext createContext(String userName, String password) {
      return null;
   }

   @Override
   public JMSContext createContext(String userName, String password, int sessionMode) {
      return null;
   }

   @Override
   public String getMessageSelector() throws JMSException {
      return null;
   }

   @Override
   public Message receive() throws JMSException {
      return null;
   }

   @Override
   public Message receive(long timeout) throws JMSException {
      final org.fusesource.mqtt.client.Message message;
      try {
         message = blockingConnection.receive(timeout, TimeUnit.MILLISECONDS);
      } catch (Exception e) {
         throw new JMSException(e.getMessage());
      }
      if (message != null) {
         return new TMessage(new String(message.getPayload()));
      }
      return null;
   }

   @Override
   public Message receiveNoWait() throws JMSException {
      return null;
   }

   @Override
   public void setDisableMessageID(boolean value) throws JMSException {

   }

   @Override
   public boolean getDisableMessageID() throws JMSException {
      return false;
   }

   @Override
   public void setDisableMessageTimestamp(boolean value) throws JMSException {

   }

   @Override
   public boolean getDisableMessageTimestamp() throws JMSException {
      return false;
   }

   @Override
   public void setDeliveryMode(int deliveryMode) throws JMSException {

   }

   @Override
   public int getDeliveryMode() throws JMSException {
      return 0;
   }

   @Override
   public void setPriority(int defaultPriority) throws JMSException {

   }

   @Override
   public int getPriority() throws JMSException {
      return 0;
   }

   @Override
   public void setTimeToLive(long timeToLive) throws JMSException {

   }

   @Override
   public long getTimeToLive() throws JMSException {
      return 0;
   }

   @Override
   public Destination getDestination() throws JMSException {
      return null;
   }

   @Override
   public void send(Message message) throws JMSException {
      try {
         blockingConnection.publish(topicName, message.getBody(String.class).getBytes(), QoS.EXACTLY_ONCE, false);
      } catch (Exception e) {
         throw new JMSException(e.getMessage());
      }
   }

   @Override
   public void send(Message message, int deliveryMode, int priority, long timeToLive) throws JMSException {

   }

   @Override
   public void send(Destination destination, Message message) throws JMSException {

   }

   @Override
   public void send(Message message, CompletionListener completionListener) throws JMSException {

   }

   @Override
   public void send(Destination destination,
                    Message message,
                    int deliveryMode,
                    int priority,
                    long timeToLive) throws JMSException {

   }

   @Override
   public void send(Destination destination,
                    Message message,
                    CompletionListener completionListener) throws JMSException {

   }

   @Override
   public void send(Message message,
                    int deliveryMode,
                    int priority,
                    long timeToLive,
                    CompletionListener completionListener) throws JMSException {

   }

   @Override
   public void send(Destination destination,
                    Message message,
                    int deliveryMode,
                    int priority,
                    long timeToLive,
                    CompletionListener completionListener) throws JMSException {

   }

   @Override
   public long getDeliveryDelay() throws JMSException {
      return 0;
   }

   @Override
   public void setDeliveryDelay(long deliveryDelay) throws JMSException {

   }

   @Override
   public BytesMessage createBytesMessage() throws JMSException {
      return null;
   }

   @Override
   public MapMessage createMapMessage() throws JMSException {
      return null;
   }

   @Override
   public Message createMessage() throws JMSException {
      return null;
   }

   @Override
   public ObjectMessage createObjectMessage() throws JMSException {
      return null;
   }

   @Override
   public ObjectMessage createObjectMessage(Serializable object) throws JMSException {
      return null;
   }

   @Override
   public StreamMessage createStreamMessage() throws JMSException {
      return null;
   }

   @Override
   public TextMessage createTextMessage() throws JMSException {
      return null;
   }

   @Override
   public TextMessage createTextMessage(String text) throws JMSException {
      return new TMessage(text);
   }

   @Override
   public boolean getTransacted() throws JMSException {
      return false;
   }

   @Override
   public int getAcknowledgeMode() throws JMSException {
      return 0;
   }

   @Override
   public void commit() throws JMSException {

   }

   @Override
   public void rollback() throws JMSException {

   }

   @Override
   public Session createSession(boolean transacted, int acknowledgeMode) throws JMSException {
      return this;
   }

   @Override
   public Session createSession(int sessionMode) throws JMSException {
      return null;
   }

   @Override
   public Session createSession() throws JMSException {
      return null;
   }

   @Override
   public String getClientID() throws JMSException {
      return null;
   }

   @Override
   public void setClientID(String clientID) throws JMSException {

   }

   @Override
   public ConnectionMetaData getMetaData() throws JMSException {
      return null;
   }

   @Override
   public ExceptionListener getExceptionListener() throws JMSException {
      return null;
   }

   @Override
   public void setExceptionListener(ExceptionListener listener) throws JMSException {

   }

   @Override
   public void start() throws JMSException {
      blockingConnection = mqtt.blockingConnection();
      try {
         blockingConnection.connect();

         if (consumer) {
            blockingConnection.subscribe(new org.fusesource.mqtt.client.Topic[]{new org.fusesource.mqtt.client.Topic(topicName, QoS.EXACTLY_ONCE)});
         }

      } catch (Exception e) {
         throw new JMSException(e.getMessage());
      }
   }

   @Override
   public void stop() throws JMSException {

   }

   @Override
   public void close() throws JMSException {

   }

   @Override
   public ConnectionConsumer createConnectionConsumer(Destination destination,
                                                      String messageSelector,
                                                      ServerSessionPool sessionPool,
                                                      int maxMessages) throws JMSException {
      return null;
   }

   @Override
   public ConnectionConsumer createDurableConnectionConsumer(Topic topic,
                                                             String subscriptionName,
                                                             String messageSelector,
                                                             ServerSessionPool sessionPool,
                                                             int maxMessages) throws JMSException {
      return null;
   }

   @Override
   public ConnectionConsumer createSharedDurableConnectionConsumer(Topic topic,
                                                                   String subscriptionName,
                                                                   String messageSelector,
                                                                   ServerSessionPool sessionPool,
                                                                   int maxMessages) throws JMSException {
      return null;
   }

   @Override
   public ConnectionConsumer createSharedConnectionConsumer(Topic topic,
                                                            String subscriptionName,
                                                            String messageSelector,
                                                            ServerSessionPool sessionPool,
                                                            int maxMessages) throws JMSException {
      return null;
   }

   @Override
   public void recover() throws JMSException {

   }

   @Override
   public MessageListener getMessageListener() throws JMSException {
      return null;
   }

   @Override
   public void setMessageListener(MessageListener listener) throws JMSException {

   }

   @Override
   public void run() {

   }

   @Override
   public MessageProducer createProducer(Destination destination) throws JMSException {
      return this;
   }

   @Override
   public MessageConsumer createConsumer(Destination destination) throws JMSException {
      consumer = true;
      return this;
   }

   @Override
   public MessageConsumer createConsumer(Destination destination, String messageSelector) throws JMSException {
      return null;
   }

   @Override
   public MessageConsumer createConsumer(Destination destination,
                                         String messageSelector,
                                         boolean NoLocal) throws JMSException {
      return null;
   }

   @Override
   public Queue createQueue(String queueName) throws JMSException {
      return null;
   }

   @Override
   public Topic createTopic(String topicName) throws JMSException {
      this.topicName = topicName;
      return this;
   }

   @Override
   public TopicSubscriber createDurableSubscriber(Topic topic, String name) throws JMSException {
      return null;
   }

   @Override
   public TopicSubscriber createDurableSubscriber(Topic topic,
                                                  String name,
                                                  String messageSelector,
                                                  boolean noLocal) throws JMSException {
      return null;
   }

   @Override
   public QueueBrowser createBrowser(Queue queue) throws JMSException {
      return null;
   }

   @Override
   public QueueBrowser createBrowser(Queue queue, String messageSelector) throws JMSException {
      return null;
   }

   @Override
   public TemporaryQueue createTemporaryQueue() throws JMSException {
      return null;
   }

   @Override
   public TemporaryTopic createTemporaryTopic() throws JMSException {
      return null;
   }

   @Override
   public void unsubscribe(String name) throws JMSException {

   }

   @Override
   public MessageConsumer createSharedConsumer(Topic topic, String sharedSubscriptionName) throws JMSException {
      return null;
   }

   @Override
   public MessageConsumer createSharedConsumer(Topic topic,
                                               String sharedSubscriptionName,
                                               String messageSelector) throws JMSException {
      return null;
   }

   @Override
   public MessageConsumer createDurableConsumer(Topic topic, String name) throws JMSException {
      return null;
   }

   @Override
   public MessageConsumer createDurableConsumer(Topic topic,
                                                String name,
                                                String messageSelector,
                                                boolean noLocal) throws JMSException {
      return null;
   }

   @Override
   public MessageConsumer createSharedDurableConsumer(Topic topic, String name) throws JMSException {
      return null;
   }

   @Override
   public MessageConsumer createSharedDurableConsumer(Topic topic,
                                                      String name,
                                                      String messageSelector) throws JMSException {
      return null;
   }

   @Override
   public String getTopicName() throws JMSException {
      return topicName;
   }

   private class TMessage implements Message, TextMessage {

      final String s;

      TMessage(String s) {
         this.s = s;
      }

      @Override
      public String getJMSMessageID() throws JMSException {
         return null;
      }

      @Override
      public void setJMSMessageID(String id) throws JMSException {

      }

      @Override
      public long getJMSTimestamp() throws JMSException {
         return 0;
      }

      @Override
      public void setJMSTimestamp(long timestamp) throws JMSException {

      }

      @Override
      public byte[] getJMSCorrelationIDAsBytes() throws JMSException {
         return new byte[0];
      }

      @Override
      public void setJMSCorrelationIDAsBytes(byte[] correlationID) throws JMSException {

      }

      @Override
      public void setJMSCorrelationID(String correlationID) throws JMSException {

      }

      @Override
      public String getJMSCorrelationID() throws JMSException {
         return null;
      }

      @Override
      public Destination getJMSReplyTo() throws JMSException {
         return null;
      }

      @Override
      public void setJMSReplyTo(Destination replyTo) throws JMSException {

      }

      @Override
      public Destination getJMSDestination() throws JMSException {
         return null;
      }

      @Override
      public void setJMSDestination(Destination destination) throws JMSException {

      }

      @Override
      public int getJMSDeliveryMode() throws JMSException {
         return 0;
      }

      @Override
      public void setJMSDeliveryMode(int deliveryMode) throws JMSException {

      }

      @Override
      public boolean getJMSRedelivered() throws JMSException {
         return false;
      }

      @Override
      public void setJMSRedelivered(boolean redelivered) throws JMSException {

      }

      @Override
      public String getJMSType() throws JMSException {
         return null;
      }

      @Override
      public void setJMSType(String type) throws JMSException {

      }

      @Override
      public long getJMSExpiration() throws JMSException {
         return 0;
      }

      @Override
      public void setJMSExpiration(long expiration) throws JMSException {

      }

      @Override
      public int getJMSPriority() throws JMSException {
         return 0;
      }

      @Override
      public void setJMSPriority(int priority) throws JMSException {

      }

      @Override
      public void clearProperties() throws JMSException {

      }

      @Override
      public boolean propertyExists(String name) throws JMSException {
         return false;
      }

      @Override
      public boolean getBooleanProperty(String name) throws JMSException {
         return false;
      }

      @Override
      public byte getByteProperty(String name) throws JMSException {
         return 0;
      }

      @Override
      public short getShortProperty(String name) throws JMSException {
         return 0;
      }

      @Override
      public int getIntProperty(String name) throws JMSException {
         return 0;
      }

      @Override
      public long getLongProperty(String name) throws JMSException {
         return 0;
      }

      @Override
      public float getFloatProperty(String name) throws JMSException {
         return 0;
      }

      @Override
      public double getDoubleProperty(String name) throws JMSException {
         return 0;
      }

      @Override
      public String getStringProperty(String name) throws JMSException {
         return null;
      }

      @Override
      public Object getObjectProperty(String name) throws JMSException {
         return null;
      }

      @Override
      public Enumeration getPropertyNames() throws JMSException {
         return null;
      }

      @Override
      public void setBooleanProperty(String name, boolean value) throws JMSException {

      }

      @Override
      public void setByteProperty(String name, byte value) throws JMSException {

      }

      @Override
      public void setShortProperty(String name, short value) throws JMSException {

      }

      @Override
      public void setIntProperty(String name, int value) throws JMSException {

      }

      @Override
      public void setLongProperty(String name, long value) throws JMSException {

      }

      @Override
      public void setFloatProperty(String name, float value) throws JMSException {

      }

      @Override
      public void setDoubleProperty(String name, double value) throws JMSException {

      }

      @Override
      public void setStringProperty(String name, String value) throws JMSException {

      }

      @Override
      public void setObjectProperty(String name, Object value) throws JMSException {

      }

      @Override
      public void acknowledge() throws JMSException {

      }

      @Override
      public void clearBody() throws JMSException {

      }

      @Override
      public long getJMSDeliveryTime() throws JMSException {
         return 0;
      }

      @Override
      public void setJMSDeliveryTime(long deliveryTime) throws JMSException {

      }

      @Override
      public <T> T getBody(Class<T> c) throws JMSException {
         return (T) s;
      }

      @Override
      public boolean isBodyAssignableTo(Class c) throws JMSException {
         return false;
      }

      @Override
      public void setText(String string) throws JMSException {
      }

      @Override
      public String getText() throws JMSException {
         return s;
      }
   }
}