diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java index ec4d38ab84..a722e0ba3e 100755 --- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java +++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java @@ -170,9 +170,10 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon private final AtomicBoolean transportFailed = new AtomicBoolean(false); private final CopyOnWriteArrayList sessions = new CopyOnWriteArrayList(); private final CopyOnWriteArrayList connectionConsumers = new CopyOnWriteArrayList(); + private final CopyOnWriteArrayList transportListeners = new CopyOnWriteArrayList(); + // Stream are deprecated and will be removed in a later release. private final CopyOnWriteArrayList inputStreams = new CopyOnWriteArrayList(); private final CopyOnWriteArrayList outputStreams = new CopyOnWriteArrayList(); - private final CopyOnWriteArrayList transportListeners = new CopyOnWriteArrayList(); // Maps ConsumerIds to ActiveMQConsumer objects private final ConcurrentHashMap dispatchers = new ConcurrentHashMap(); @@ -677,6 +678,7 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon ActiveMQConnectionConsumer c = i.next(); c.dispose(); } + // Stream are deprecated and will be removed in a later release. for (Iterator i = this.inputStreams.iterator(); i.hasNext();) { ActiveMQInputStream c = i.next(); c.dispose(); @@ -1598,6 +1600,8 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon ActiveMQConnectionConsumer c = i.next(); c.dispose(); } + + // Stream are deprecated and will be removed in a later release. for (Iterator i = this.inputStreams.iterator(); i.hasNext();) { ActiveMQInputStream c = i.next(); c.dispose(); @@ -2186,45 +2190,54 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon } @Override + @Deprecated public InputStream createInputStream(Destination dest) throws JMSException { return createInputStream(dest, null); } @Override + @Deprecated public InputStream createInputStream(Destination dest, String messageSelector) throws JMSException { return createInputStream(dest, messageSelector, false); } @Override + @Deprecated public InputStream createInputStream(Destination dest, String messageSelector, boolean noLocal) throws JMSException { return createInputStream(dest, messageSelector, noLocal, -1); } @Override + @Deprecated public InputStream createInputStream(Destination dest, String messageSelector, boolean noLocal, long timeout) throws JMSException { return doCreateInputStream(dest, messageSelector, noLocal, null, timeout); } @Override + @Deprecated public InputStream createDurableInputStream(Topic dest, String name) throws JMSException { return createInputStream(dest, null, false); } @Override + @Deprecated public InputStream createDurableInputStream(Topic dest, String name, String messageSelector) throws JMSException { return createDurableInputStream(dest, name, messageSelector, false); } @Override + @Deprecated public InputStream createDurableInputStream(Topic dest, String name, String messageSelector, boolean noLocal) throws JMSException { return createDurableInputStream(dest, name, messageSelector, noLocal, -1); } @Override + @Deprecated public InputStream createDurableInputStream(Topic dest, String name, String messageSelector, boolean noLocal, long timeout) throws JMSException { return doCreateInputStream(dest, messageSelector, noLocal, name, timeout); } + @Deprecated private InputStream doCreateInputStream(Destination dest, String messageSelector, boolean noLocal, String subName, long timeout) throws JMSException { checkClosedOrFailed(); ensureConnectionInfoSent(); @@ -2236,6 +2249,7 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon * to disk/database by the broker */ @Override + @Deprecated public OutputStream createOutputStream(Destination dest) throws JMSException { return createOutputStream(dest, null, ActiveMQMessage.DEFAULT_DELIVERY_MODE, ActiveMQMessage.DEFAULT_PRIORITY, ActiveMQMessage.DEFAULT_TIME_TO_LIVE); } @@ -2244,6 +2258,7 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon * Creates a non persistent output stream; messages will not be written to * disk */ + @Deprecated public OutputStream createNonPersistentOutputStream(Destination dest) throws JMSException { return createOutputStream(dest, null, DeliveryMode.NON_PERSISTENT, ActiveMQMessage.DEFAULT_PRIORITY, ActiveMQMessage.DEFAULT_TIME_TO_LIVE); } @@ -2261,6 +2276,7 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon * method */ @Override + @Deprecated public OutputStream createOutputStream(Destination dest, Map streamProperties, int deliveryMode, int priority, long timeToLive) throws JMSException { checkClosedOrFailed(); ensureConnectionInfoSent(); @@ -2338,18 +2354,22 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon } } + @Deprecated public void addOutputStream(ActiveMQOutputStream stream) { outputStreams.add(stream); } + @Deprecated public void removeOutputStream(ActiveMQOutputStream stream) { outputStreams.remove(stream); } + @Deprecated public void addInputStream(ActiveMQInputStream stream) { inputStreams.add(stream); } + @Deprecated public void removeInputStream(ActiveMQInputStream stream) { inputStreams.remove(stream); } diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQInputStream.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQInputStream.java index 4937431248..a8954217fa 100644 --- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQInputStream.java +++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQInputStream.java @@ -43,6 +43,7 @@ import org.apache.activemq.util.JMSExceptionSupport; /** * */ +@Deprecated public class ActiveMQInputStream extends InputStream implements ActiveMQDispatcher { private final ActiveMQConnection connection; diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQOutputStream.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQOutputStream.java index 0427c9dfbd..c6e400d2ba 100644 --- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQOutputStream.java +++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQOutputStream.java @@ -36,6 +36,7 @@ import org.apache.activemq.util.IntrospectionSupport; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +@Deprecated public class ActiveMQOutputStream extends OutputStream implements Disposable { private static final Logger LOG = LoggerFactory.getLogger(ActiveMQOutputStream.class); diff --git a/activemq-client/src/main/java/org/apache/activemq/StreamConnection.java b/activemq-client/src/main/java/org/apache/activemq/StreamConnection.java index 3684719409..2f75b4f39d 100644 --- a/activemq-client/src/main/java/org/apache/activemq/StreamConnection.java +++ b/activemq-client/src/main/java/org/apache/activemq/StreamConnection.java @@ -31,9 +31,10 @@ import javax.jms.Topic; * Destination in using standard java InputStream and OutputStream objects. It's * best use case is to send and receive large amounts of data that would be to * large to hold in a single JMS message. - * - * + * + * */ +@Deprecated public interface StreamConnection extends Connection { InputStream createInputStream(Destination dest) throws JMSException; @@ -49,7 +50,7 @@ public interface StreamConnection extends Connection { InputStream createDurableInputStream(Topic dest, String name, String messageSelector) throws JMSException; InputStream createDurableInputStream(Topic dest, String name, String messageSelector, boolean noLocal) throws JMSException; - + InputStream createDurableInputStream(Topic dest, String name, String messageSelector, boolean noLocal, long timeout) throws JMSException; OutputStream createOutputStream(Destination dest) throws JMSException; @@ -67,7 +68,7 @@ public interface StreamConnection extends Connection { * TopicSubscriber for the subscription, or while a consumed * message is part of a pending transaction or has not been acknowledged in * the session. - * + * * @param name the name used to identify this subscription * @throws JMSException if the session fails to unsubscribe to the durable * subscription due to some internal error. diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/ActiveMQInputStreamTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/ActiveMQInputStreamTest.java index ce628ad910..77f422e159 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/ActiveMQInputStreamTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/ActiveMQInputStreamTest.java @@ -23,11 +23,13 @@ import javax.jms.Queue; import javax.jms.Session; import junit.framework.TestCase; + import org.apache.activemq.broker.BrokerService; import org.apache.activemq.command.ActiveMQDestination; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +@Deprecated public class ActiveMQInputStreamTest extends TestCase { private static final Logger LOG = LoggerFactory.getLogger(ActiveMQInputStreamTest.class); @@ -39,6 +41,7 @@ public class ActiveMQInputStreamTest extends TestCase { private BrokerService broker; private String connectionUri; + @Override public void setUp() throws Exception { broker = new BrokerService(); broker.setUseJmx(false); @@ -53,6 +56,7 @@ public class ActiveMQInputStreamTest extends TestCase { connectionUri = broker.getTransportConnectors().get(0).getPublishableConnectString(); } + @Override public void tearDown() throws Exception { broker.stop(); broker.waitUntilStopped(); diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/streams/JMSInputStreamTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/streams/JMSInputStreamTest.java index 25b514a834..f3926628f1 100755 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/streams/JMSInputStreamTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/streams/JMSInputStreamTest.java @@ -41,6 +41,7 @@ import org.apache.activemq.command.ActiveMQTopic; /** * JMSInputStreamTest */ +@Deprecated public class JMSInputStreamTest extends JmsTestSupport { public Destination destination;