ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From akuznet...@apache.org
Subject [35/46] incubator-ignite git commit: ignite-534 - IgniteJmsStreamer implemented. Closes #14.
Date Tue, 18 Aug 2015 16:12:24 GMT
ignite-534 - IgniteJmsStreamer implemented. Closes #14.


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/0d701127
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/0d701127
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/0d701127

Branch: refs/heads/ignite-843
Commit: 0d701127c8a95a30f4c484247e84ab1f10f95e3a
Parents: d0536dc
Author: Raul Kripalani <raulk@apache.org>
Authored: Fri Aug 14 16:11:19 2015 -0700
Committer: Valentin Kulichenko <valentin.kulichenko@gmail.com>
Committed: Fri Aug 14 16:11:19 2015 -0700

----------------------------------------------------------------------
 .../ignite/internal/util/GridArgumentCheck.java |  17 +
 modules/jms11/pom.xml                           |  98 ++++
 .../apache/ignite/stream/jms11/JmsStreamer.java | 524 ++++++++++++++++++
 .../ignite/stream/jms11/MessageTransformer.java |  43 ++
 .../stream/jms11/IgniteJmsStreamerTest.java     | 551 +++++++++++++++++++
 .../jms11/IgniteJmsStreamerTestSuite.java       |  34 ++
 .../ignite/stream/jms11/TestTransformers.java   | 123 +++++
 pom.xml                                         |   1 +
 8 files changed, 1391 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0d701127/modules/core/src/main/java/org/apache/ignite/internal/util/GridArgumentCheck.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/GridArgumentCheck.java b/modules/core/src/main/java/org/apache/ignite/internal/util/GridArgumentCheck.java
index b94bae6..8289f90 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/GridArgumentCheck.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/GridArgumentCheck.java
@@ -35,6 +35,9 @@ public class GridArgumentCheck {
     /** Not empty argument error message suffix. */
     private static final String NOT_EMPTY_SUFFIX = " must not be empty.";
 
+    /** Not null or empty error message suffix. */
+    private static final String NOT_NULL_OR_EMPTY_SUFFIX = " must not be null or empty.";
+
     /**
      * Checks if given argument value is not {@code null}. Otherwise - throws {@link NullPointerException}.
      *
@@ -145,4 +148,18 @@ public class GridArgumentCheck {
         if (arr.length == 0)
             throw new IllegalArgumentException(INVALID_ARG_MSG_PREFIX + name + NOT_EMPTY_SUFFIX);
     }
+
+    /**
+     * Checks that a String is not null or empty.
+     *
+     * @param value Value to check.
+     * @param name Argument name.
+     */
+    public static void notNullOrEmpty(String value, String name) {
+        notNull(value, name);
+
+        if (value.trim().length() == 0) {
+            throw new IllegalArgumentException(INVALID_ARG_MSG_PREFIX + name + NOT_NULL_OR_EMPTY_SUFFIX);
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0d701127/modules/jms11/pom.xml
----------------------------------------------------------------------
diff --git a/modules/jms11/pom.xml b/modules/jms11/pom.xml
new file mode 100644
index 0000000..c93ea9a
--- /dev/null
+++ b/modules/jms11/pom.xml
@@ -0,0 +1,98 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+  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.
+-->
+
+<!--
+    POM file.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.apache.ignite</groupId>
+        <artifactId>ignite-parent</artifactId>
+        <version>1</version>
+        <relativePath>../../parent</relativePath>
+    </parent>
+
+    <artifactId>ignite-jms11</artifactId>
+    <version>1.4.1-SNAPSHOT</version>
+
+    <properties>
+        <activemq.version>5.11.1</activemq.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>javax.jms</groupId>
+            <artifactId>jms-api</artifactId>
+            <version>1.1-rev-1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.activemq</groupId>
+            <artifactId>activemq-client</artifactId>
+            <version>${activemq.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.activemq</groupId>
+            <artifactId>activemq-broker</artifactId>
+            <version>${activemq.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.activemq</groupId>
+            <artifactId>activemq-kahadb-store</artifactId>
+            <version>${activemq.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-log4j</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-spring</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.ignite</groupId>
+            <artifactId>ignite-core</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0d701127/modules/jms11/src/main/java/org/apache/ignite/stream/jms11/JmsStreamer.java
----------------------------------------------------------------------
diff --git a/modules/jms11/src/main/java/org/apache/ignite/stream/jms11/JmsStreamer.java b/modules/jms11/src/main/java/org/apache/ignite/stream/jms11/JmsStreamer.java
new file mode 100644
index 0000000..de84a3e
--- /dev/null
+++ b/modules/jms11/src/main/java/org/apache/ignite/stream/jms11/JmsStreamer.java
@@ -0,0 +1,524 @@
+/*
+ * 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.ignite.stream.jms11;
+
+import org.apache.ignite.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.stream.*;
+
+import javax.jms.*;
+import javax.jms.Queue;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+
+/**
+ * Streamer that consumes from a JMS destination and feeds key-value pairs into an {@link IgniteDataStreamer} instance.
+ * <p>
+ * This Streamer uses purely JMS semantics and it is not coupled with any JMS implementation. It uses {@link
+ * MessageListener} to receive messages. You must provide your broker's {@link javax.jms.ConnectionFactory} when
+ * creating a {@link JmsStreamer}.
+ * <p>
+ * You must also provide a {@link MessageTransformer} to convert the incoming message into cache entries.
+ * <p>
+ * This Streamer has many features:
+ *
+ * <ul>
+ *     <li>Consumes from queues or topics.</li>
+ *     <li>For topics, it supports durable subscriptions.</li>
+ *     <li>Concurrent consumers are supported via the <tt>threads</tt> parameter. When consuming from queues,
+ *     this component will start as many {@link Session} objects with separate {@link MessageListener} instances each,
+ *     therefore achieving <i>native</i> concurrency (in terms of the JMS standard).<br>
+ *     When consuming from topics, obviously we cannot start multiple threads as that would lead us to consume
+ *     duplicate messages. Therefore, we achieve concurrency in a <i>virtualized</i> manner through an internal
+ *     thread pool.</li>
+ *     <li>Transacted sessions are supported through the <tt>transacted</tt> parameter.</li>
+ *     <li>Batched consumption is possible via the <tt>batched</tt> parameter. Depending on the broker, this
+ *     technique can provide a higher throughput as it decreases the amount of message acknowledgement round trips
+ *     that are necessary, albeit at the expense possible duplicate messages (especially if an incident
+ *     occurs in the middle of a transaction).<br>
+ *     Batches are committed when the <tt>batchClosureMillis</tt> time has elapsed, or when a Session has received
+ *     at least <tt>batchClosureSize</tt> messages. Time-based closure fires with the specified frequency and applies to
+ *     all {@link Session}s in parallel. Size-based closure applies individually to each <tt>Session</tt> (as transactions
+ *     are <tt>Session-bound</tt> in JMS, so it will fire when that {@link Session} has processed that many messages.
+ *     Both options are compatible with each other, or you can disable either (see setter documentation),
+ *     but not both.</li>
+ *     <li>Can specify the destination with implementation-specific {@link Destination} objects or with names.</li>
+ * </ul>
+
+ *
+ * @author Raul Kripalani
+ */
+public class JmsStreamer<T extends Message, K, V> extends StreamAdapter<T, K, V> {
+
+    /** Logger. */
+    private IgniteLogger log;
+
+    /**
+     * <i>Compulsory.</i> The message transformer that converts an incoming JMS {@link Message} (or subclass) into one
+     * or multiple cache entries.
+     */
+    private MessageTransformer<T, K, V> transformer;
+
+    /** The JMS {@link ConnectionFactory} to use. */
+    private ConnectionFactory connectionFactory;
+
+    /** Whether to register or not as a durable subscription (for topic consumption). */
+    private boolean durableSubscription;
+
+    /** Name of the durable subscription, as required by the JMS specification. */
+    private String durableSubscriptionName;
+
+    /** Client ID in case we're using durable subscribers. */
+    private String clientId;
+
+    /** The JMS {@link Destination}; takes precedence over destinationName if both are set. */
+    private Destination destination;
+
+    /** Name of the destination. */
+    private String destinationName;
+
+    /** Whether to consume in a transacted manner. */
+    private boolean transacted;
+
+    /** Whether to consume messages in batches. May lead to duplicate consumption. Value <tt>true</tt> implies
+     * <tt>transacted = true</tt>. */
+    private boolean batched;
+
+    /** When using batched consumers, the amount of messages after the batch (transaction) will be committed. */
+    private int batchClosureSize = 50;
+
+    /**
+     * When using batched consumers, the amount of time to wait before the batch (transaction) will be committed. A
+     * value of 0 or -1 disables timed-based session commits.
+     */
+    private long batchClosureMillis = 1000;
+
+    /** Destination type. */
+    private Class<? extends Destination> destinationType = Queue.class;
+
+    /**
+     * Number of threads to concurrently consume JMS messages. When working with queues, we will start as many {@link
+     * javax.jms.Session} objects as indicated by this field, i.e. you will get native concurrency. On the other hand,
+     * when consuming from a topic, for obvious reason we will only start 1 message consumer but we will distribute the
+     * processing of received messages to as many concurrent threads as indicated.
+     */
+    private int threads = 1;
+
+    /** Whether we are stopped or not. */
+    private volatile boolean stopped = true;
+
+    /** JMS Connection. */
+    private Connection connection;
+
+    /** Stores the current JMS Sessions. */
+    private Set<Session> sessions = Collections.newSetFromMap(new ConcurrentHashMap<Session, Boolean>());
+
+    /** Message consumers. */
+    private Set<MessageConsumer> consumers = Collections.newSetFromMap(new ConcurrentHashMap<MessageConsumer, Boolean>());
+
+    /** Message listeners. */
+    private Set<IgniteJmsMessageListener> listeners = Collections.newSetFromMap(new ConcurrentHashMap<IgniteJmsMessageListener, Boolean>());
+
+    /** Scheduler for handling {@link #batchClosureMillis}. */
+    private ScheduledExecutorService scheduler;
+
+    /**
+     * Starts streamer.
+     *
+     * @throws IgniteException If failed.
+     */
+    public void start() throws IgniteException {
+        if (!stopped)
+            throw new IgniteException("Attempted to start an already started JMS Streamer");
+
+        try {
+            A.notNull(getStreamer(), "streamer");
+            A.notNull(getIgnite(), "ignite");
+
+            log = getIgnite().log();
+
+            A.notNull(transformer, "message transformer");
+            A.notNull(connectionFactory, "connection factory");
+            A.ensure(threads > 0, "threads > 0");
+
+            // handle batched && transacted parameter interaction
+            if (batched && !transacted) {
+                log.warning("Starting a Batched JMS Streamer without transacted flag = true. Setting it automatically.");
+                transacted = true;
+            }
+
+            // handle batch completion criteria
+            if (batched) {
+                A.ensure(batchClosureMillis > 0 || batchClosureSize > 0, "at least one of batch closure size or " +
+                    "batch closure frequency must be specified when using batch consumption");
+            }
+
+            // check the parameters needed for durable subscriptions, if enabled
+            if (durableSubscription) {
+                A.notNullOrEmpty(clientId, "client id is compulsory when using durable subscriptions");
+                A.notNullOrEmpty(durableSubscriptionName, "durable subscription name is compulsory when using " +
+                    "durable subscriptions");
+            }
+
+            // validate the destination; if we have an explicit destination, make sure it's of type Queue or Topic;
+            // else make sure that the destinationName and the destinationType are valid
+            if (destination == null) {
+                A.notNull(destinationType, "destination type");
+                A.ensure(destinationType.isAssignableFrom(Queue.class) || destinationType.isAssignableFrom(Topic.class),
+                    "this streamer can only handle Queues or Topics.");
+                A.notNullOrEmpty(destinationName, "destination or destination name");
+            }
+            else if (destination instanceof Queue) {
+                destinationType = Queue.class;
+            }
+            else if (destination instanceof Topic) {
+                destinationType = Topic.class;
+            }
+            else {
+                throw new IllegalArgumentException("Invalid destination object. Can only handle Queues or Topics.");
+            }
+
+            // create a new connection and the client iD if relevant.
+            connection = connectionFactory.createConnection();
+            if (clientId != null && clientId.trim().length() > 0) {
+                connection.setClientID(clientId.trim());
+            }
+
+            // build the JMS objects
+            if (destinationType == Queue.class) {
+                initializeJmsObjectsForQueue();
+            }
+            else {
+                initializeJmsObjectsForTopic();
+            }
+
+            stopped = false;
+
+            // start the JMS connection
+            connection.start();
+
+            // set up the scheduler service for committing batches
+            if (batched && batchClosureMillis > 0) {
+                scheduler = Executors.newScheduledThreadPool(1);
+                scheduler.schedule(new Runnable() {
+                    @Override
+                    public void run() {
+                        for (Session session : sessions) {
+                            try {
+                                session.commit();
+                                if (log.isDebugEnabled()) {
+                                    log.debug("Committing session from time-based batch completion [session=" +
+                                        session + "]");
+                                }
+                            }
+                            catch (JMSException e) {
+                                log.warning("Error while committing session: from batch time-based completion " +
+                                    "[session=" + session + "]");
+                            }
+                        }
+                        for (IgniteJmsMessageListener ml : listeners) {
+                            ml.resetBatchCounter();
+                        }
+                    }
+                }, batchClosureMillis, TimeUnit.MILLISECONDS);
+            }
+
+        }
+        catch (Throwable t) {
+            throw new IgniteException("Exception while initializing JmsStreamer", t);
+        }
+
+    }
+
+    /**
+     * Stops streamer.
+     */
+    public void stop() throws IgniteException {
+        if (stopped)
+            throw new IgniteException("Attempted to stop an already stopped JMS Streamer");
+
+
+        try {
+            stopped = true;
+
+            if (scheduler != null && !scheduler.isShutdown()) {
+                scheduler.shutdown();
+                scheduler = null;
+            }
+
+            connection.stop();
+            connection.close();
+
+            for (Session s : sessions) {
+                s.close();
+            }
+
+            sessions.clear();
+            consumers.clear();
+            listeners.clear();
+        }
+        catch (Throwable t) {
+            throw new IgniteException("Exception while stopping JmsStreamer", t);
+        }
+    }
+
+    /**
+     * Sets the JMS {@link ConnectionFactory}.
+     *
+     * @param connectionFactory JMS {@link ConnectionFactory} for this streamer to use.
+     */
+    public void setConnectionFactory(ConnectionFactory connectionFactory) {
+        this.connectionFactory = connectionFactory;
+    }
+
+    /**
+     * <i>Compulsory.</i> The {@link MessageTransformer} that converts an incoming JMS {@link Message} (or subclass)
+     * into one or multiple cache entries.
+     *
+     * @param transformer The implementation of the MessageTransformer to use.
+     */
+    public void setTransformer(MessageTransformer<T, K, V> transformer) {
+        this.transformer = transformer;
+    }
+
+    /**
+     * Sets the JMS {@link Destination} explicitly. Takes precedence over destinationName if both are set.
+     *
+     * @param destination JMS {@link Destination} if setting it explicitly.
+     */
+    public void setDestination(Destination destination) {
+        this.destination = destination;
+    }
+
+    /**
+     * Sets the name of the JMS destination to consume from.
+     *
+     * @param destinationName The name of the destination; will be passed on directly to the broker.
+     */
+    public void setDestinationName(String destinationName) {
+        this.destinationName = destinationName;
+    }
+
+    /**
+     * Sets the type of the destination to create, when used in combination with {@link #setDestinationName(String)}. It
+     * can be an interface or the implementation class specific to the broker.
+     *
+     * @param destinationType The class representing the destination type. Suggested values: {@link Queue} or {@link
+     * Topic}. <i>Compulsory</i> if using {@link #destinationName}.
+     * @see Queue
+     * @see Topic
+     */
+    public void setDestinationType(Class<? extends Destination> destinationType) {
+        this.destinationType = destinationType;
+    }
+
+    /**
+     * Sets the number of threads to concurrently consume JMS messages. <p> When working with queues, we will start as
+     * many {@link javax.jms.Session} objects as indicated by this field, i.e. you will get native concurrency. <p> On
+     * the other hand, when consuming from a topic, for obvious reason we will only start 1 message consumer but we will
+     * distribute the processing of received messages to as many concurrent threads as indicated.
+     *
+     * @param threads Number of threads to use. Default: <tt>1</tt>.
+     */
+    public void setThreads(int threads) {
+        this.threads = threads;
+    }
+
+    /**
+     * Sets the client ID of the JMS {@link Connection}.
+     *
+     * @param clientId Client ID in case we're using durable subscribers. Default: none.
+     */
+    public void setClientId(String clientId) {
+        this.clientId = clientId;
+    }
+
+    /**
+     * A <tt>true</tt> value is only accepted in combination with topic consumption.
+     *
+     * @param durableSubscription Whether or not to use durable subscriptions. Default: <tt>false</tt>.
+     */
+    public void setDurableSubscription(boolean durableSubscription) {
+        this.durableSubscription = durableSubscription;
+    }
+
+    /**
+     * Instructs the streamer whether to use local JMS transactions or not.
+     *
+     * @param transacted Whether to consume or not in a transacted manner. Default: <tt>false</tt>.
+     */
+    public void setTransacted(boolean transacted) {
+        this.transacted = transacted;
+    }
+
+    /**
+     * Batch consumption leverages JMS Transactions to minimise round trips to the broker. <p> Rather than ACKing every
+     * single message received, they will be received in the context of a JMS transaction which will be committed once
+     * the indicated batch closure size or batch closure time has elapsed. <p> Warning: May lead to duplicate
+     * consumption.
+     *
+     * @param batched Whether to consume messages in batches. Value <tt>true</tt> implies <tt>transacted = true</tt>.
+     * Default: <tt>false</tt>.
+     * @see #setBatchClosureMillis(long)
+     * @see #setBatchClosureSize(int)
+     */
+    public void setBatched(boolean batched) {
+        this.batched = batched;
+    }
+
+    /**
+     * When using batched consumption, sets the amount of messages that will be received before a batch is committed.
+     *
+     * @param batchClosureSize The amount of messages processed before a batch is committed. Default: <tt>50</tt>.
+     */
+    public void setBatchClosureSize(int batchClosureSize) {
+        this.batchClosureSize = batchClosureSize;
+    }
+
+    /**
+     * When using batched consumption, sets the time in milliseconds that will elapse before a batch is committed.
+     *
+     * @param batchClosureMillis Milliseconds before a batch is committed. Default: <tt>1000ms</tt>.
+     */
+    public void setBatchClosureMillis(long batchClosureMillis) {
+        this.batchClosureMillis = batchClosureMillis;
+    }
+
+    /**
+     * When using Durable Subscribers, sets the name of the durable subscriber. It is compulsory.
+     *
+     * @param durableSubscriptionName Name of the durable subscriber. Default: none.
+     */
+    public void setDurableSubscriptionName(String durableSubscriptionName) {
+        this.durableSubscriptionName = durableSubscriptionName;
+    }
+
+    private void initializeJmsObjectsForTopic() throws JMSException {
+        Session session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
+        Topic topic = (Topic)destination;
+
+        if (destination == null)
+            topic = session.createTopic(destinationName);
+
+        MessageConsumer consumer = durableSubscription ? session.createDurableSubscriber(topic, durableSubscriptionName) :
+            session.createConsumer(topic);
+
+        IgniteJmsMessageListener messageListener = new IgniteJmsMessageListener(session, true);
+        consumer.setMessageListener(messageListener);
+
+        consumers.add(consumer);
+        sessions.add(session);
+        listeners.add(messageListener);
+    }
+
+    private void initializeJmsObjectsForQueue() throws JMSException {
+        for (int i = 0; i < threads; i++) {
+            Session session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
+
+            if (destination == null)
+                destination = session.createQueue(destinationName);
+
+            MessageConsumer consumer = session.createConsumer(destination);
+
+            IgniteJmsMessageListener messageListener = new IgniteJmsMessageListener(session, false);
+            consumer.setMessageListener(messageListener);
+
+            consumers.add(consumer);
+            sessions.add(session);
+            listeners.add(messageListener);
+        }
+    }
+
+    private void processMessage(T message) {
+        final IgniteDataStreamer<K, V> streamer = getStreamer();
+
+        Map<K, V> entries = transformer.apply(message);
+
+        if (entries == null || entries.size() == 0)
+            return;
+
+        streamer.addData(entries);
+    }
+
+    /**
+     * Message listener for queues.
+     */
+    private class IgniteJmsMessageListener implements MessageListener {
+
+        private Session session;
+        private AtomicInteger counter = new AtomicInteger(0);
+        private Executor executor;
+
+        public IgniteJmsMessageListener(Session session, boolean createThreadPool) {
+            this.session = session;
+
+            // if we don't need a thread pool, create a dummy one that executes the task synchronously
+            //noinspection NullableProblems
+            this.executor = createThreadPool ? Executors.newFixedThreadPool(threads) : new Executor() {
+                @Override
+                public void execute(Runnable command) {
+                    command.run();
+                }
+            };
+        }
+
+        @Override
+        public void onMessage(final Message message) {
+            if (stopped) {
+                return;
+            }
+
+            executor.execute(new Runnable() {
+                @Override @SuppressWarnings("unchecked")
+                public void run() {
+                    processMessage((T)message);
+                    if (batched) {
+                        // batch completion may be handled by timer only
+                        if (batchClosureSize <= 0)
+                            return;
+
+                        else if (counter.incrementAndGet() >= batchClosureSize) {
+                            try {
+                                session.commit();
+                                counter.set(0);
+                            }
+                            catch (Exception e) {
+                                log.warning("Could not commit JMS session upon completion of batch.", e);
+                            }
+                        }
+                    }
+                    else if (transacted) {
+                        try {
+                            session.commit();
+                        }
+                        catch (JMSException e) {
+                            log.warning("Could not commit JMS session (non-batched).", e);
+                        }
+                    }
+                }
+            });
+
+        }
+
+        public void resetBatchCounter() {
+            counter.set(0);
+        }
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0d701127/modules/jms11/src/main/java/org/apache/ignite/stream/jms11/MessageTransformer.java
----------------------------------------------------------------------
diff --git a/modules/jms11/src/main/java/org/apache/ignite/stream/jms11/MessageTransformer.java b/modules/jms11/src/main/java/org/apache/ignite/stream/jms11/MessageTransformer.java
new file mode 100644
index 0000000..3cb0231
--- /dev/null
+++ b/modules/jms11/src/main/java/org/apache/ignite/stream/jms11/MessageTransformer.java
@@ -0,0 +1,43 @@
+/*
+ * 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.ignite.stream.jms11;
+
+import org.jetbrains.annotations.*;
+
+import javax.jms.*;
+import java.util.*;
+
+/**
+ * Implement this interface to transform from a {@link Message} to a set of cache entries in the form of a {@link Map}.
+ *
+ * @param <T> The type of JMS Message.
+ * @param <K> The type of the cache key.
+ * @param <V> The type of the cache value.
+ * @author Raul Kripalani
+ */
+public interface MessageTransformer<T extends Message, K, V> {
+
+    /**
+     * Transformation function.
+     *
+     * @param message The message received from the JMS broker.
+     * @return Set of cache entries to add to the cache. It could be empty or null if the message should be skipped.
+     */
+    @Nullable Map<K, V> apply(T message);
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0d701127/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/IgniteJmsStreamerTest.java
----------------------------------------------------------------------
diff --git a/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/IgniteJmsStreamerTest.java b/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/IgniteJmsStreamerTest.java
new file mode 100644
index 0000000..f1c107d
--- /dev/null
+++ b/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/IgniteJmsStreamerTest.java
@@ -0,0 +1,551 @@
+/*
+ * 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.ignite.stream.jms11;
+
+import org.apache.activemq.*;
+import org.apache.activemq.broker.*;
+import org.apache.activemq.broker.region.*;
+import org.apache.activemq.broker.region.policy.*;
+import org.apache.activemq.command.*;
+import org.apache.ignite.*;
+import org.apache.ignite.events.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.testframework.junits.common.*;
+
+import org.junit.*;
+
+import javax.jms.*;
+import javax.jms.Destination;
+import javax.jms.Message;
+import javax.jms.Queue;
+import javax.jms.Topic;
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.apache.ignite.events.EventType.*;
+
+/**
+ * Test for {@link JmsStreamer}. Tests both queues and topics.
+ *
+ * @author Raul Kripalani
+ */
+public class IgniteJmsStreamerTest extends GridCommonAbstractTest {
+
+    private static final int CACHE_ENTRY_COUNT = 100;
+    private static final String QUEUE_NAME = "ignite.test.queue";
+    private static final String TOPIC_NAME = "ignite.test.topic";
+    private static final Map<String, String> TEST_DATA = new HashMap<>();
+
+    static {
+        for (int i = 1; i <= CACHE_ENTRY_COUNT; i++)
+            TEST_DATA.put(Integer.toString(i), "v" + i);
+    }
+
+    private BrokerService broker;
+    private ConnectionFactory connectionFactory;
+
+    /** Constructor. */
+    public IgniteJmsStreamerTest() {
+        super(true);
+    }
+
+    @Before @SuppressWarnings("unchecked")
+    public void beforeTest() throws Exception {
+        grid().<Integer, String>getOrCreateCache(defaultCacheConfiguration());
+
+        broker = new BrokerService();
+        broker.deleteAllMessages();
+        broker.setPersistent(false);
+
+        PolicyMap policyMap = new PolicyMap();
+        PolicyEntry policy = new PolicyEntry();
+        policy.setQueuePrefetch(1);
+        broker.setDestinationPolicy(policyMap);
+        broker.getDestinationPolicy().setDefaultEntry(policy);
+
+        broker.start(true);
+
+        connectionFactory = new ActiveMQConnectionFactory(BrokerRegistry.getInstance().findFirst().getVmConnectorURI());
+
+    }
+
+    @After
+    public void afterTest() throws Exception {
+        grid().cache(null).clear();
+
+        broker.deleteAllMessages();
+        broker.stop();
+    }
+
+    public void testQueueFromName() throws Exception {
+        Destination destination = new ActiveMQQueue(QUEUE_NAME);
+
+        // produce messages into the queue
+        produceObjectMessages(destination, false);
+
+        try (IgniteDataStreamer<String, String> dataStreamer = grid().dataStreamer(null)) {
+            JmsStreamer<ObjectMessage, String, String> jmsStreamer = newJmsStreamer(ObjectMessage.class, dataStreamer);
+            jmsStreamer.setDestinationType(Queue.class);
+            jmsStreamer.setDestinationName(QUEUE_NAME);
+
+            // subscribe to cache PUT events and return a countdown latch starting at CACHE_ENTRY_COUNT
+            CountDownLatch latch = subscribeToPutEvents(CACHE_ENTRY_COUNT);
+
+            jmsStreamer.start();
+
+            // all cache PUT events received in 10 seconds
+            latch.await(10, TimeUnit.SECONDS);
+
+            assertAllCacheEntriesLoaded();
+
+            jmsStreamer.stop();
+        }
+
+    }
+
+    public void testTopicFromName() throws JMSException, InterruptedException {
+        Destination destination = new ActiveMQTopic(TOPIC_NAME);
+
+        // should not produced messages until subscribed to the topic; otherwise they will be missed because this is not
+        // a durable subscriber (for which a dedicated test exists)
+
+        try (IgniteDataStreamer<String, String> dataStreamer = grid().dataStreamer(null)) {
+            JmsStreamer<ObjectMessage, String, String> jmsStreamer = newJmsStreamer(ObjectMessage.class, dataStreamer);
+            jmsStreamer.setDestinationType(Topic.class);
+            jmsStreamer.setDestinationName(TOPIC_NAME);
+
+            // subscribe to cache PUT events and return a countdown latch starting at CACHE_ENTRY_COUNT
+            CountDownLatch latch = subscribeToPutEvents(CACHE_ENTRY_COUNT);
+
+            jmsStreamer.start();
+
+            // produce messages
+            produceObjectMessages(destination, false);
+
+            // all cache PUT events received in 10 seconds
+            latch.await(10, TimeUnit.SECONDS);
+
+            assertAllCacheEntriesLoaded();
+
+            jmsStreamer.stop();
+        }
+
+    }
+
+    public void testQueueFromExplicitDestination() throws Exception {
+        Destination destination = new ActiveMQQueue(QUEUE_NAME);
+
+        // produce messages into the queue
+        produceObjectMessages(destination, false);
+
+        try (IgniteDataStreamer<String, String> dataStreamer = grid().dataStreamer(null)) {
+            JmsStreamer<ObjectMessage, String, String> jmsStreamer = newJmsStreamer(ObjectMessage.class, dataStreamer);
+            jmsStreamer.setDestination(destination);
+
+            // subscribe to cache PUT events and return a countdown latch starting at CACHE_ENTRY_COUNT
+            CountDownLatch latch = subscribeToPutEvents(CACHE_ENTRY_COUNT);
+
+            // start the streamer
+            jmsStreamer.start();
+
+            // all cache PUT events received in 10 seconds
+            latch.await(10, TimeUnit.SECONDS);
+
+            assertAllCacheEntriesLoaded();
+
+            jmsStreamer.stop();
+        }
+
+    }
+
+    public void testTopicFromExplicitDestination() throws JMSException, InterruptedException {
+        Destination destination = new ActiveMQTopic(TOPIC_NAME);
+
+        // should not produced messages until subscribed to the topic; otherwise they will be missed because this is not
+        // a durable subscriber (for which a dedicated test exists)
+
+        try (IgniteDataStreamer<String, String> dataStreamer = grid().dataStreamer(null)) {
+            JmsStreamer<ObjectMessage, String, String> jmsStreamer = newJmsStreamer(ObjectMessage.class, dataStreamer);
+            jmsStreamer.setDestination(destination);
+
+            // subscribe to cache PUT events and return a countdown latch starting at CACHE_ENTRY_COUNT
+            CountDownLatch latch = subscribeToPutEvents(CACHE_ENTRY_COUNT);
+
+            jmsStreamer.start();
+
+            // produce messages
+            produceObjectMessages(destination, false);
+
+            // all cache PUT events received in 10 seconds
+            latch.await(10, TimeUnit.SECONDS);
+
+            assertAllCacheEntriesLoaded();
+
+            jmsStreamer.stop();
+        }
+
+    }
+
+    public void testInsertMultipleCacheEntriesFromOneMessage() throws Exception {
+        Destination destination = new ActiveMQQueue(QUEUE_NAME);
+
+        // produce A SINGLE MESSAGE, containing all data, into the queue
+        produceStringMessages(destination, true);
+
+        try (IgniteDataStreamer<String, String> dataStreamer = grid().dataStreamer(null)) {
+            JmsStreamer<TextMessage, String, String> jmsStreamer = newJmsStreamer(TextMessage.class, dataStreamer);
+            jmsStreamer.setDestination(destination);
+
+            // subscribe to cache PUT events and return a countdown latch starting at CACHE_ENTRY_COUNT
+            CountDownLatch latch = subscribeToPutEvents(CACHE_ENTRY_COUNT);
+
+            jmsStreamer.start();
+
+            // all cache PUT events received in 10 seconds
+            latch.await(10, TimeUnit.SECONDS);
+
+            assertAllCacheEntriesLoaded();
+
+            jmsStreamer.stop();
+        }
+
+    }
+
+    public void testDurableSubscriberStartStopStart() throws Exception {
+        Destination destination = new ActiveMQTopic(TOPIC_NAME);
+
+        try (IgniteDataStreamer<String, String> dataStreamer = grid().dataStreamer(null)) {
+            JmsStreamer<TextMessage, String, String> jmsStreamer = newJmsStreamer(TextMessage.class, dataStreamer);
+            jmsStreamer.setDestination(destination);
+            jmsStreamer.setDurableSubscription(true);
+            jmsStreamer.setClientId(Long.toString(System.currentTimeMillis()));
+            jmsStreamer.setDurableSubscriptionName("ignite-test-durable");
+
+            // we start the streamer so that the durable subscriber registers itself
+            jmsStreamer.start();
+
+            // we stop it immediately
+            jmsStreamer.stop();
+
+            // we assert that there are no clients of the broker (to make sure we disconnected properly)
+            assertEquals(0, broker.getCurrentConnections());
+
+            // we send messages while we're still away
+            produceStringMessages(destination, false);
+
+            // subscribe to cache PUT events and return a countdown latch starting at CACHE_ENTRY_COUNT
+            CountDownLatch latch = subscribeToPutEvents(CACHE_ENTRY_COUNT);
+
+            jmsStreamer.start();
+
+            // all cache PUT events received in 10 seconds
+            latch.await(10, TimeUnit.SECONDS);
+
+            assertAllCacheEntriesLoaded();
+
+            jmsStreamer.stop();
+        }
+
+    }
+
+    public void testQueueMessagesConsumedInBatchesCompletionSizeBased() throws Exception {
+        Destination destination = new ActiveMQQueue(QUEUE_NAME);
+
+        // produce multiple messages into the queue
+        produceStringMessages(destination, false);
+
+        try (IgniteDataStreamer<String, String> dataStreamer = grid().dataStreamer(null)) {
+            JmsStreamer<TextMessage, String, String> jmsStreamer = newJmsStreamer(TextMessage.class, dataStreamer);
+            jmsStreamer.setDestination(destination);
+            jmsStreamer.setBatched(true);
+            jmsStreamer.setBatchClosureSize(99);
+
+            // disable time-based session commits
+            jmsStreamer.setBatchClosureMillis(0);
+
+            // subscribe to cache PUT events and return a countdown latch starting at CACHE_ENTRY_COUNT
+            CountDownLatch latch = subscribeToPutEvents(CACHE_ENTRY_COUNT);
+
+            jmsStreamer.start();
+
+            // all cache PUT events received in 10 seconds
+            latch.await(10, TimeUnit.SECONDS);
+
+            assertAllCacheEntriesLoaded();
+
+            // we expect all entries to be loaded, but still one (uncommitted) message should remain in the queue
+            // as observed by the broker
+            DestinationStatistics qStats = broker.getBroker().getDestinationMap().get(destination).getDestinationStatistics();
+            assertEquals(1, qStats.getMessages().getCount());
+            assertEquals(1, qStats.getInflight().getCount());
+
+            jmsStreamer.stop();
+        }
+
+    }
+
+    public void testQueueMessagesConsumedInBatchesCompletionTimeBased() throws Exception {
+        Destination destination = new ActiveMQQueue(QUEUE_NAME);
+
+        // produce multiple messages into the queue
+        produceStringMessages(destination, false);
+
+        try (IgniteDataStreamer<String, String> dataStreamer = grid().dataStreamer(null)) {
+            JmsStreamer<TextMessage, String, String> jmsStreamer = newJmsStreamer(TextMessage.class, dataStreamer);
+            jmsStreamer.setDestination(destination);
+            jmsStreamer.setBatched(true);
+            jmsStreamer.setBatchClosureMillis(2000);
+            // disable size-based session commits
+            jmsStreamer.setBatchClosureSize(0);
+
+            // subscribe to cache PUT events and return a countdown latch starting at CACHE_ENTRY_COUNT
+            CountDownLatch latch = subscribeToPutEvents(CACHE_ENTRY_COUNT);
+            DestinationStatistics qStats = broker.getBroker().getDestinationMap().get(destination).getDestinationStatistics();
+
+            jmsStreamer.start();
+
+            // all messages are still inflight
+            assertEquals(CACHE_ENTRY_COUNT, qStats.getMessages().getCount());
+            assertEquals(0, qStats.getDequeues().getCount());
+
+            // wait a little bit
+            Thread.sleep(100);
+
+            // all messages are still inflight
+            assertEquals(CACHE_ENTRY_COUNT, qStats.getMessages().getCount());
+            assertEquals(0, qStats.getDequeues().getCount());
+
+            // now let the scheduler execute
+            Thread.sleep(2100);
+
+            // all messages are committed
+            assertEquals(0, qStats.getMessages().getCount());
+            assertEquals(CACHE_ENTRY_COUNT, qStats.getDequeues().getCount());
+
+            latch.await(5, TimeUnit.SECONDS);
+
+            assertAllCacheEntriesLoaded();
+
+            jmsStreamer.stop();
+        }
+
+    }
+
+    public void testGenerateNoEntries() throws Exception {
+        Destination destination = new ActiveMQQueue(QUEUE_NAME);
+
+        // produce multiple messages into the queue
+        produceStringMessages(destination, false);
+
+        try (IgniteDataStreamer<String, String> dataStreamer = grid().dataStreamer(null)) {
+            JmsStreamer<TextMessage, String, String> jmsStreamer = newJmsStreamer(TextMessage.class, dataStreamer);
+            // override the transformer with one that generates no cache entries
+            jmsStreamer.setTransformer(TestTransformers.generateNoEntries());
+            jmsStreamer.setDestination(destination);
+
+            // subscribe to cache PUT events and return a countdown latch starting at CACHE_ENTRY_COUNT
+            CountDownLatch latch = subscribeToPutEvents(1);
+
+            jmsStreamer.start();
+
+            // no cache PUT events were received in 3 seconds, i.e. CountDownLatch does not fire
+            assertFalse(latch.await(3, TimeUnit.SECONDS));
+
+            jmsStreamer.stop();
+        }
+
+    }
+
+    public void testTransactedSessionNoBatching() throws Exception {
+        Destination destination = new ActiveMQQueue(QUEUE_NAME);
+
+        // produce multiple messages into the queue
+        produceStringMessages(destination, false);
+
+        try (IgniteDataStreamer<String, String> dataStreamer = grid().dataStreamer(null)) {
+            JmsStreamer<TextMessage, String, String> jmsStreamer = newJmsStreamer(TextMessage.class, dataStreamer);
+            jmsStreamer.setTransacted(true);
+            jmsStreamer.setDestination(destination);
+
+            // subscribe to cache PUT events and return a countdown latch starting at CACHE_ENTRY_COUNT
+            CountDownLatch latch = subscribeToPutEvents(CACHE_ENTRY_COUNT);
+
+            jmsStreamer.start();
+
+            // all cache PUT events received in 10 seconds
+            latch.await(10, TimeUnit.SECONDS);
+
+            assertAllCacheEntriesLoaded();
+
+            jmsStreamer.stop();
+        }
+
+    }
+
+    public void testQueueMultipleThreads() throws Exception {
+        Destination destination = new ActiveMQQueue(QUEUE_NAME);
+
+        // produce messages into the queue
+        produceObjectMessages(destination, false);
+
+        try (IgniteDataStreamer<String, String> dataStreamer = grid().dataStreamer(null)) {
+            JmsStreamer<ObjectMessage, String, String> jmsStreamer = newJmsStreamer(ObjectMessage.class, dataStreamer);
+            jmsStreamer.setDestination(destination);
+            jmsStreamer.setThreads(5);
+
+            // subscribe to cache PUT events and return a countdown latch starting at CACHE_ENTRY_COUNT
+            CountDownLatch latch = subscribeToPutEvents(CACHE_ENTRY_COUNT);
+
+            // start the streamer
+            jmsStreamer.start();
+
+            DestinationStatistics qStats = broker.getBroker().getDestinationMap().get(destination).getDestinationStatistics();
+            assertEquals(5, qStats.getConsumers().getCount());
+
+            // all cache PUT events received in 10 seconds
+            latch.await(10, TimeUnit.SECONDS);
+
+            // assert that all consumers received messages - given that the prefetch is 1
+            for (Subscription subscription : broker.getBroker().getDestinationMap().get(destination).getConsumers())
+                assertTrue(subscription.getDequeueCounter() > 0);
+
+            assertAllCacheEntriesLoaded();
+
+            jmsStreamer.stop();
+        }
+
+    }
+
+    private void assertAllCacheEntriesLoaded() {
+        // Get the cache and check that the entries are present
+        IgniteCache<String, String> cache = grid().cache(null);
+        for (Map.Entry<String, String> entry : TEST_DATA.entrySet())
+            assertEquals(entry.getValue(), cache.get(entry.getKey()));
+    }
+
+    @SuppressWarnings("unchecked")
+    private <T extends Message> JmsStreamer<T, String, String> newJmsStreamer(Class<T> type,
+        IgniteDataStreamer<String, String> dataStreamer) {
+
+        JmsStreamer<T, String, String> jmsStreamer = new JmsStreamer<>();
+        jmsStreamer.setIgnite(grid());
+        jmsStreamer.setStreamer(dataStreamer);
+        jmsStreamer.setConnectionFactory(connectionFactory);
+
+        if (type == ObjectMessage.class) {
+            jmsStreamer.setTransformer((MessageTransformer<T, String, String>) TestTransformers.forObjectMessage());
+        }
+        else {
+            jmsStreamer.setTransformer((MessageTransformer<T, String, String>) TestTransformers.forTextMessage());
+        }
+
+        dataStreamer.allowOverwrite(true);
+        dataStreamer.autoFlushFrequency(10);
+        return jmsStreamer;
+    }
+
+    private CountDownLatch subscribeToPutEvents(int expect) {
+        Ignite ignite = grid();
+
+        // Listen to cache PUT events and expect as many as messages as test data items
+        final CountDownLatch latch = new CountDownLatch(expect);
+        @SuppressWarnings("serial") IgniteBiPredicate<UUID, CacheEvent> callback = new IgniteBiPredicate<UUID, CacheEvent>() {
+            @Override public boolean apply(UUID uuid, CacheEvent evt) {
+                latch.countDown();
+                return true;
+            }
+        };
+
+        ignite.events(ignite.cluster().forCacheNodes(null)).remoteListen(callback, null, EVT_CACHE_OBJECT_PUT);
+        return latch;
+    }
+
+    private void produceObjectMessages(Destination destination, boolean singleMessage) throws JMSException {
+        Session session = connectionFactory.createConnection().createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageProducer mp = session.createProducer(destination);
+        HashSet<TestTransformers.TestObject> set = new HashSet<>();
+
+        for (String key : TEST_DATA.keySet()) {
+            TestTransformers.TestObject to = new TestTransformers.TestObject(key, TEST_DATA.get(key));
+            set.add(to);
+        }
+
+        int messagesSent;
+        if (singleMessage) {
+            mp.send(session.createObjectMessage(set));
+            messagesSent = 1;
+        }
+        else {
+            for (TestTransformers.TestObject to : set)
+                mp.send(session.createObjectMessage(to));
+
+            messagesSent = set.size();
+        }
+
+        if (destination instanceof Queue) {
+            try {
+                assertEquals(messagesSent, broker.getBroker().getDestinationMap().get(destination)
+                    .getDestinationStatistics().getMessages().getCount());
+            }
+            catch (Exception e) {
+                fail(e.toString());
+            }
+        }
+
+    }
+
+    private void produceStringMessages(Destination destination, boolean singleMessage) throws JMSException {
+        Session session = connectionFactory.createConnection().createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageProducer mp = session.createProducer(destination);
+        HashSet<String> set = new HashSet<>();
+
+        for (String key : TEST_DATA.keySet())
+            set.add(key + "," + TEST_DATA.get(key));
+
+        int messagesSent;
+        if (singleMessage) {
+            StringBuilder sb = new StringBuilder();
+
+            for (String s : set)
+                sb.append(s).append("|");
+
+            sb.deleteCharAt(sb.length() - 1);
+            mp.send(session.createTextMessage(sb.toString()));
+            messagesSent = 1;
+
+        }
+        else {
+            for (String s : set) {
+                mp.send(session.createTextMessage(s));
+            }
+            messagesSent = set.size();
+        }
+
+        if (destination instanceof Queue) {
+            try {
+                assertEquals(messagesSent, broker.getBroker().getDestinationMap().get(destination)
+                    .getDestinationStatistics().getMessages().getCount());
+            }
+            catch (Exception e) {
+                fail(e.toString());
+            }
+        }
+
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0d701127/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/IgniteJmsStreamerTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/IgniteJmsStreamerTestSuite.java b/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/IgniteJmsStreamerTestSuite.java
new file mode 100644
index 0000000..e299f04
--- /dev/null
+++ b/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/IgniteJmsStreamerTestSuite.java
@@ -0,0 +1,34 @@
+/*
+ * 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.ignite.stream.jms11;
+
+import org.junit.runner.*;
+import org.junit.runners.*;
+
+/**
+ * JMS streamer tests.
+ *
+ * @author Raul Kripalani
+ */
+@RunWith(Suite.class)
+@Suite.SuiteClasses({
+    IgniteJmsStreamerTest.class
+})
+public class IgniteJmsStreamerTestSuite {
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0d701127/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/TestTransformers.java
----------------------------------------------------------------------
diff --git a/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/TestTransformers.java b/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/TestTransformers.java
new file mode 100644
index 0000000..ce001a1
--- /dev/null
+++ b/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/TestTransformers.java
@@ -0,0 +1,123 @@
+/*
+ * 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.ignite.stream.jms11;
+
+import javax.jms.*;
+import java.io.*;
+import java.util.*;
+
+/**
+ * Test transformers for JmsStreamer tests.
+ *
+ * @author Raul Kripalani
+ */
+public class TestTransformers {
+
+    /**
+     * Returns a transformer for JMS {@link TextMessage}s, capable of extracting many tuples from a single message,
+     * if pipe characters are encountered.
+     *
+     * @return
+     */
+    public static MessageTransformer<TextMessage, String, String> forTextMessage() {
+        return new MessageTransformer<TextMessage, String, String>() {
+            @Override
+            public Map<String, String> apply(TextMessage message) {
+                final Map<String, String> answer = new HashMap<>();
+                String text;
+                try {
+                    text = message.getText();
+                }
+                catch (JMSException e) {
+                    e.printStackTrace();
+                    return Collections.emptyMap();
+                }
+                for (String s : text.split("\\|")) {
+                    String[] tokens = s.split(",");
+                    answer.put(tokens[0], tokens[1]);
+                }
+                return answer;
+            }
+        };
+    }
+
+    /**
+     * Returns a transformer for JMS {@link ObjectMessage}s, capable of extracting many tuples from a single message,
+     * if the payload is a {@link Collection}.
+     *
+     * @return
+     */
+    public static MessageTransformer<ObjectMessage, String, String> forObjectMessage() {
+        return new MessageTransformer<ObjectMessage, String, String>() {
+            @Override @SuppressWarnings("unchecked")
+            public Map<String, String> apply(ObjectMessage message) {
+                Object object;
+                try {
+                    object = message.getObject();
+                }
+                catch (JMSException e) {
+                    e.printStackTrace();
+                    return Collections.emptyMap();
+                }
+
+                final Map<String, String> answer = new HashMap<>();
+                if (object instanceof Collection) {
+                    for (TestObject to : (Collection<TestObject>)object)
+                        answer.put(to.getKey(), to.getValue());
+
+                }
+                else if (object instanceof TestObject) {
+                    TestObject to = (TestObject)object;
+                    answer.put(to.getKey(), to.getValue());
+                }
+                return answer;
+            }
+        };
+    }
+
+    public static MessageTransformer<TextMessage, String, String> generateNoEntries() {
+        return new MessageTransformer<TextMessage, String, String>() {
+            @Override
+            public Map<String, String> apply(TextMessage message) {
+                return null;
+            }
+        };
+    }
+
+    public static class TestObject implements Serializable {
+        private static final long serialVersionUID = -7332027566186690945L;
+
+        private String key;
+        private String value;
+
+        public TestObject(String key, String value) {
+            this.key = key;
+            this.value = value;
+        }
+
+        public String getKey() {
+            return key;
+        }
+
+        public String getValue() {
+            return value;
+        }
+
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0d701127/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 4d53d96..f3a5d65 100644
--- a/pom.xml
+++ b/pom.xml
@@ -74,6 +74,7 @@
         <module>modules/mesos</module>
         <module>modules/kafka</module>
         <module>modules/yarn</module>
+        <module>modules/jms11</module>
     </modules>
 
     <profiles>


Mime
View raw message