You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2015/12/07 19:39:51 UTC
[39/42] activemq-artemis git commit: Add missing @Override annotations
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionFactoryImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionFactoryImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionFactoryImpl.java
index 66d5cbb..1a67dff 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionFactoryImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionFactoryImpl.java
@@ -227,10 +227,12 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
connectionReadyForWrites = true;
}
+ @Override
public void disableFinalizeCheck() {
finalizeCheck = false;
}
+ @Override
public Lock lockFailover() {
newFailoverLock.lock();
return newFailoverLock;
@@ -244,6 +246,7 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
}
}
+ @Override
public void connect(final int initialConnectAttempts,
final boolean failoverOnInitialConnection) throws ActiveMQException {
// Get the connection
@@ -259,10 +262,12 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
}
+ @Override
public TransportConfiguration getConnectorConfiguration() {
return connectorConfig;
}
+ @Override
public void setBackupConnector(final TransportConfiguration live, final TransportConfiguration backUp) {
Connector localConnector = connector;
@@ -290,10 +295,12 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
}
}
+ @Override
public Object getBackupConnector() {
return backupConfig;
}
+ @Override
public ClientSession createSession(final String username,
final String password,
final boolean xa,
@@ -304,35 +311,42 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
return createSessionInternal(username, password, xa, autoCommitSends, autoCommitAcks, preAcknowledge, ackBatchSize);
}
+ @Override
public ClientSession createSession(final boolean autoCommitSends,
final boolean autoCommitAcks,
final int ackBatchSize) throws ActiveMQException {
return createSessionInternal(null, null, false, autoCommitSends, autoCommitAcks, serverLocator.isPreAcknowledge(), ackBatchSize);
}
+ @Override
public ClientSession createXASession() throws ActiveMQException {
return createSessionInternal(null, null, true, false, false, serverLocator.isPreAcknowledge(), serverLocator.getAckBatchSize());
}
+ @Override
public ClientSession createTransactedSession() throws ActiveMQException {
return createSessionInternal(null, null, false, false, false, serverLocator.isPreAcknowledge(), serverLocator.getAckBatchSize());
}
+ @Override
public ClientSession createSession() throws ActiveMQException {
return createSessionInternal(null, null, false, true, true, serverLocator.isPreAcknowledge(), serverLocator.getAckBatchSize());
}
+ @Override
public ClientSession createSession(final boolean autoCommitSends,
final boolean autoCommitAcks) throws ActiveMQException {
return createSessionInternal(null, null, false, autoCommitSends, autoCommitAcks, serverLocator.isPreAcknowledge(), serverLocator.getAckBatchSize());
}
+ @Override
public ClientSession createSession(final boolean xa,
final boolean autoCommitSends,
final boolean autoCommitAcks) throws ActiveMQException {
return createSessionInternal(null, null, xa, autoCommitSends, autoCommitAcks, serverLocator.isPreAcknowledge(), serverLocator.getAckBatchSize());
}
+ @Override
public ClientSession createSession(final boolean xa,
final boolean autoCommitSends,
final boolean autoCommitAcks,
@@ -342,11 +356,13 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
// ConnectionLifeCycleListener implementation --------------------------------------------------
+ @Override
public void connectionCreated(final ActiveMQComponent component,
final Connection connection,
final String protocol) {
}
+ @Override
public void connectionDestroyed(final Object connectionID) {
// The exception has to be created in the same thread where it's being called
// as to avoid a different stack trace cause
@@ -355,6 +371,7 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
// It has to use the same executor as the disconnect message is being sent through
closeExecutor.execute(new Runnable() {
+ @Override
public void run() {
handleConnectionFailure(connectionID, ex);
}
@@ -362,18 +379,21 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
}
+ @Override
public void connectionException(final Object connectionID, final ActiveMQException me) {
handleConnectionFailure(connectionID, me);
}
// Must be synchronized to prevent it happening concurrently with failover which can lead to
// inconsistencies
+ @Override
public void removeSession(final ClientSessionInternal session, final boolean failingOver) {
synchronized (sessions) {
sessions.remove(session);
}
}
+ @Override
public void connectionReadyForWrites(final Object connectionID, final boolean ready) {
synchronized (connectionReadyLock) {
if (connectionReadyForWrites != ready) {
@@ -385,31 +405,38 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
}
}
+ @Override
public synchronized int numConnections() {
return connection != null ? 1 : 0;
}
+ @Override
public int numSessions() {
return sessions.size();
}
+ @Override
public void addFailureListener(final SessionFailureListener listener) {
listeners.add(listener);
}
+ @Override
public boolean removeFailureListener(final SessionFailureListener listener) {
return listeners.remove(listener);
}
+ @Override
public ClientSessionFactoryImpl addFailoverListener(FailoverEventListener listener) {
failoverListeners.add(listener);
return this;
}
+ @Override
public boolean removeFailoverListener(FailoverEventListener listener) {
return failoverListeners.remove(listener);
}
+ @Override
public void causeExit() {
clientProtocolManager.stop();
}
@@ -447,6 +474,7 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
checkCloseConnection();
}
+ @Override
public void close() {
if (closed) {
return;
@@ -456,6 +484,7 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
serverLocator.factoryClosed(this);
}
+ @Override
public void cleanup() {
if (closed) {
return;
@@ -464,6 +493,7 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
interruptConnectAndCloseAllSessions(false);
}
+ @Override
public boolean isClosed() {
return closed || serverLocator.isClosed();
}
@@ -856,6 +886,7 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
}
}
+ @Override
public RemotingConnection getConnection() {
if (closed)
throw new IllegalStateException("ClientSessionFactory is closed!");
@@ -948,6 +979,7 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
// else... we will try to instantiate a new one
return AccessController.doPrivileged(new PrivilegedAction<ConnectorFactory>() {
+ @Override
public ConnectorFactory run() {
return (ConnectorFactory) ClassloadingUtil.newInstanceFromClassLoader(connectorFactoryClassName);
}
@@ -966,6 +998,7 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
// Must be executed on new thread since cannot block the Netty thread for a long time and fail
// can cause reconnect loop
+ @Override
public void run() {
try {
CLOSE_RUNNABLES.add(this);
@@ -990,10 +1023,12 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
}
+ @Override
public void setReconnectAttempts(final int attempts) {
reconnectAttempts = attempts;
}
+ @Override
public Object getConnector() {
return connector;
}
@@ -1117,6 +1152,7 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
private class DelegatingBufferHandler implements BufferHandler {
+ @Override
public void bufferReceived(final Object connectionID, final ActiveMQBuffer buffer) {
RemotingConnection theConn = connection;
@@ -1162,6 +1198,7 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
pingRunnable = new WeakReference<PingRunnable>(runnable);
}
+ @Override
public void run() {
PingRunnable runnable = pingRunnable.get();
@@ -1180,6 +1217,7 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
private long lastCheck = System.currentTimeMillis();
+ @Override
public synchronized void run() {
if (cancelled || stopPingingAfterOne && !first) {
return;
@@ -1200,6 +1238,7 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
threadPool.execute(new Runnable() {
// Must be executed on different thread
+ @Override
public void run() {
connection.fail(me);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionImpl.java
index 0a24022..2942b8c 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionImpl.java
@@ -234,28 +234,33 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
// ClientSession implementation
// -----------------------------------------------------------------
+ @Override
public void createQueue(final SimpleString address, final SimpleString queueName) throws ActiveMQException {
internalCreateQueue(address, queueName, null, false, false);
}
+ @Override
public void createQueue(final SimpleString address,
final SimpleString queueName,
final boolean durable) throws ActiveMQException {
internalCreateQueue(address, queueName, null, durable, false);
}
+ @Override
public void createQueue(final String address,
final String queueName,
final boolean durable) throws ActiveMQException {
createQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName), durable);
}
+ @Override
public void createSharedQueue(SimpleString address,
SimpleString queueName,
boolean durable) throws ActiveMQException {
createSharedQueue(address, queueName, null, durable);
}
+ @Override
public void createSharedQueue(SimpleString address,
SimpleString queueName,
SimpleString filterString,
@@ -273,6 +278,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
+ @Override
public void createQueue(final SimpleString address,
final SimpleString queueName,
final SimpleString filterString,
@@ -280,6 +286,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
internalCreateQueue(address, queueName, filterString, durable, false);
}
+ @Override
public void createQueue(final String address,
final String queueName,
final String filterString,
@@ -287,26 +294,31 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
createQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName), SimpleString.toSimpleString(filterString), durable);
}
+ @Override
public void createTemporaryQueue(final SimpleString address, final SimpleString queueName) throws ActiveMQException {
internalCreateQueue(address, queueName, null, false, true);
}
+ @Override
public void createTemporaryQueue(final String address, final String queueName) throws ActiveMQException {
internalCreateQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName), null, false, true);
}
+ @Override
public void createTemporaryQueue(final SimpleString address,
final SimpleString queueName,
final SimpleString filter) throws ActiveMQException {
internalCreateQueue(address, queueName, filter, false, true);
}
+ @Override
public void createTemporaryQueue(final String address,
final String queueName,
final String filter) throws ActiveMQException {
internalCreateQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName), SimpleString.toSimpleString(filter), false, true);
}
+ @Override
public void deleteQueue(final SimpleString queueName) throws ActiveMQException {
checkClosed();
@@ -319,10 +331,12 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public void deleteQueue(final String queueName) throws ActiveMQException {
deleteQueue(SimpleString.toSimpleString(queueName));
}
+ @Override
public QueueQuery queueQuery(final SimpleString queueName) throws ActiveMQException {
checkClosed();
@@ -336,50 +350,60 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
+ @Override
public AddressQuery addressQuery(final SimpleString address) throws ActiveMQException {
checkClosed();
return sessionContext.addressQuery(address);
}
+ @Override
public ClientConsumer createConsumer(final SimpleString queueName) throws ActiveMQException {
return createConsumer(queueName, null, false);
}
+ @Override
public ClientConsumer createConsumer(final String queueName) throws ActiveMQException {
return createConsumer(SimpleString.toSimpleString(queueName));
}
+ @Override
public ClientConsumer createConsumer(final SimpleString queueName,
final SimpleString filterString) throws ActiveMQException {
return createConsumer(queueName, filterString, consumerWindowSize, consumerMaxRate, false);
}
+ @Override
public void createQueue(final String address, final String queueName) throws ActiveMQException {
createQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName));
}
+ @Override
public ClientConsumer createConsumer(final String queueName, final String filterString) throws ActiveMQException {
return createConsumer(SimpleString.toSimpleString(queueName), SimpleString.toSimpleString(filterString));
}
+ @Override
public ClientConsumer createConsumer(final SimpleString queueName,
final SimpleString filterString,
final boolean browseOnly) throws ActiveMQException {
return createConsumer(queueName, filterString, consumerWindowSize, consumerMaxRate, browseOnly);
}
+ @Override
public ClientConsumer createConsumer(final SimpleString queueName,
final boolean browseOnly) throws ActiveMQException {
return createConsumer(queueName, null, consumerWindowSize, consumerMaxRate, browseOnly);
}
+ @Override
public ClientConsumer createConsumer(final String queueName,
final String filterString,
final boolean browseOnly) throws ActiveMQException {
return createConsumer(SimpleString.toSimpleString(queueName), SimpleString.toSimpleString(filterString), browseOnly);
}
+ @Override
public ClientConsumer createConsumer(final String queueName, final boolean browseOnly) throws ActiveMQException {
return createConsumer(SimpleString.toSimpleString(queueName), null, browseOnly);
}
@@ -394,6 +418,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
* the client during that period, so failover won't occur. If we want direct consumers we need to
* rethink how they work.
*/
+ @Override
public ClientConsumer createConsumer(final SimpleString queueName,
final SimpleString filterString,
final int windowSize,
@@ -402,6 +427,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
return internalCreateConsumer(queueName, filterString, windowSize, maxRate, browseOnly);
}
+ @Override
public ClientConsumer createConsumer(final String queueName,
final String filterString,
final int windowSize,
@@ -410,18 +436,22 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
return createConsumer(SimpleString.toSimpleString(queueName), SimpleString.toSimpleString(filterString), windowSize, maxRate, browseOnly);
}
+ @Override
public ClientProducer createProducer() throws ActiveMQException {
return createProducer((SimpleString) null);
}
+ @Override
public ClientProducer createProducer(final SimpleString address) throws ActiveMQException {
return createProducer(address, producerMaxRate);
}
+ @Override
public ClientProducer createProducer(final String address) throws ActiveMQException {
return createProducer(SimpleString.toSimpleString(address));
}
+ @Override
public ClientProducer createProducer(final SimpleString address, final int maxRate) throws ActiveMQException {
return internalCreateProducer(address, maxRate);
}
@@ -430,6 +460,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
return createProducer(SimpleString.toSimpleString(address), rate);
}
+ @Override
public XAResource getXAResource() {
return this;
}
@@ -444,6 +475,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
throw ActiveMQClientMessageBundle.BUNDLE.txOutcomeUnknown();
}
+ @Override
public void commit() throws ActiveMQException {
checkClosed();
@@ -490,14 +522,17 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
workDone = false;
}
+ @Override
public boolean isRollbackOnly() {
return rollbackOnly;
}
+ @Override
public void rollback() throws ActiveMQException {
rollback(false);
}
+ @Override
public void rollback(final boolean isLastMessageAsDelivered) throws ActiveMQException {
if (ActiveMQClientLogger.LOGGER.isTraceEnabled()) {
ActiveMQClientLogger.LOGGER.trace("calling rollback(isLastMessageAsDelivered=" + isLastMessageAsDelivered + ")");
@@ -532,10 +567,12 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
rollbackOnly = false;
}
+ @Override
public void markRollbackOnly() {
rollbackOnly = true;
}
+ @Override
public ClientMessage createMessage(final byte type,
final boolean durable,
final long expiration,
@@ -544,34 +581,42 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
return new ClientMessageImpl(type, durable, expiration, timestamp, priority, initialMessagePacketSize);
}
+ @Override
public ClientMessage createMessage(final byte type, final boolean durable) {
return this.createMessage(type, durable, 0, System.currentTimeMillis(), (byte) 4);
}
+ @Override
public ClientMessage createMessage(final boolean durable) {
return this.createMessage((byte) 0, durable);
}
+ @Override
public boolean isClosed() {
return closed;
}
+ @Override
public boolean isAutoCommitSends() {
return autoCommitSends;
}
+ @Override
public boolean isAutoCommitAcks() {
return autoCommitAcks;
}
+ @Override
public boolean isBlockOnAcknowledge() {
return blockOnAcknowledge;
}
+ @Override
public boolean isXA() {
return xa;
}
+ @Override
public void resetIfNeeded() throws ActiveMQException {
if (rollbackOnly) {
ActiveMQClientLogger.LOGGER.resettingSessionAfterFailure();
@@ -579,6 +624,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public ClientSessionImpl start() throws ActiveMQException {
checkClosed();
@@ -595,6 +641,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
return this;
}
+ @Override
public void stop() throws ActiveMQException {
stop(true);
}
@@ -613,26 +660,32 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public void addFailureListener(final SessionFailureListener listener) {
sessionFactory.addFailureListener(listener);
}
+ @Override
public boolean removeFailureListener(final SessionFailureListener listener) {
return sessionFactory.removeFailureListener(listener);
}
+ @Override
public void addFailoverListener(FailoverEventListener listener) {
sessionFactory.addFailoverListener(listener);
}
+ @Override
public boolean removeFailoverListener(FailoverEventListener listener) {
return sessionFactory.removeFailoverListener(listener);
}
+ @Override
public int getVersion() {
return sessionContext.getServerVersion();
}
+ @Override
public boolean isClosing() {
return inClose;
}
@@ -650,10 +703,12 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
// ClientSessionInternal implementation
// ------------------------------------------------------------
+ @Override
public int getMinLargeMessageSize() {
return minLargeMessageSize;
}
+ @Override
public boolean isCompressLargeMessages() {
return compressLargeMessages;
}
@@ -661,10 +716,12 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
/**
* @return the cacheLargeMessageClient
*/
+ @Override
public boolean isCacheLargeMessageClient() {
return cacheLargeMessageClient;
}
+ @Override
public String getName() {
return name;
}
@@ -672,6 +729,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
/**
* Acknowledges all messages received by the consumer so far.
*/
+ @Override
public void acknowledge(final ClientConsumer consumer, final Message message) throws ActiveMQException {
// if we're pre-acknowledging then we don't need to do anything
if (preAcknowledge) {
@@ -692,6 +750,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public void individualAcknowledge(final ClientConsumer consumer, final Message message) throws ActiveMQException {
// if we're pre-acknowledging then we don't need to do anything
if (preAcknowledge) {
@@ -710,6 +769,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public void expire(final ClientConsumer consumer, final Message message) throws ActiveMQException {
checkClosed();
@@ -719,30 +779,35 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public void addConsumer(final ClientConsumerInternal consumer) {
synchronized (consumers) {
consumers.put(consumer.getConsumerContext(), consumer);
}
}
+ @Override
public void addProducer(final ClientProducerInternal producer) {
synchronized (producers) {
producers.add(producer);
}
}
+ @Override
public void removeConsumer(final ClientConsumerInternal consumer) throws ActiveMQException {
synchronized (consumers) {
consumers.remove(consumer.getConsumerContext());
}
}
+ @Override
public void removeProducer(final ClientProducerInternal producer) {
synchronized (producers) {
producers.remove(producer);
}
}
+ @Override
public void handleReceiveMessage(final ConsumerContext consumerID,
final ClientMessageInternal message) throws Exception {
ClientConsumerInternal consumer = getConsumer(consumerID);
@@ -752,6 +817,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public void handleReceiveLargeMessage(final ConsumerContext consumerID,
ClientLargeMessageInternal clientLargeMessage,
long largeMessageSize) throws Exception {
@@ -762,6 +828,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public void handleReceiveContinuation(final ConsumerContext consumerID,
byte[] chunk,
int flowControlSize,
@@ -792,6 +859,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public void close() throws ActiveMQException {
if (closed) {
ActiveMQClientLogger.LOGGER.debug("Session was already closed, giving up now, this=" + this);
@@ -821,6 +889,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
doCleanup(false);
}
+ @Override
public synchronized void cleanUp(boolean failingOver) throws ActiveMQException {
if (closed) {
return;
@@ -833,11 +902,13 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
doCleanup(failingOver);
}
+ @Override
public ClientSessionImpl setSendAcknowledgementHandler(final SendAcknowledgementHandler handler) {
sessionContext.setSendAcknowledgementHandler(handler);
return this;
}
+ @Override
public void preHandleFailover(RemotingConnection connection) {
// We lock the channel to prevent any packets to be added to the re-send
// cache during the failover process
@@ -847,6 +918,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
// Needs to be synchronized to prevent issues with occurring concurrently with close()
+ @Override
public void handleFailover(final RemotingConnection backupConnection, ActiveMQException cause) {
synchronized (this) {
if (closed) {
@@ -947,6 +1019,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
+ @Override
public void addMetaData(String key, String data) throws ActiveMQException {
synchronized (metadata) {
metadata.put(key, data);
@@ -955,14 +1028,17 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
sessionContext.addSessionMetadata(key, data);
}
+ @Override
public void addUniqueMetaData(String key, String data) throws ActiveMQException {
sessionContext.addUniqueMetaData(key, data);
}
+ @Override
public ClientSessionFactory getSessionFactory() {
return sessionFactory;
}
+ @Override
public void setAddress(final Message message, final SimpleString address) {
if (defaultAddress == null) {
defaultAddress = address;
@@ -979,48 +1055,58 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public void setPacketSize(final int packetSize) {
if (packetSize > this.initialMessagePacketSize) {
this.initialMessagePacketSize = (int) (packetSize * 1.2);
}
}
+ @Override
public void workDone() {
workDone = true;
}
+ @Override
public void sendProducerCreditsMessage(final int credits, final SimpleString address) {
sessionContext.sendProducerCreditsMessage(credits, address);
}
+ @Override
public synchronized ClientProducerCredits getCredits(final SimpleString address, final boolean anon) {
ClientProducerCredits credits = producerCreditManager.getCredits(address, anon, sessionContext);
return credits;
}
+ @Override
public void returnCredits(final SimpleString address) {
producerCreditManager.returnCredits(address);
}
+ @Override
public void handleReceiveProducerCredits(final SimpleString address, final int credits) {
producerCreditManager.receiveCredits(address, credits);
}
+ @Override
public void handleReceiveProducerFailCredits(final SimpleString address, int credits) {
producerCreditManager.receiveFailCredits(address, credits);
}
+ @Override
public ClientProducerCreditManager getProducerCreditManager() {
return producerCreditManager;
}
+ @Override
public void startCall() {
if (concurrentCall.incrementAndGet() > 1) {
ActiveMQClientLogger.LOGGER.invalidConcurrentSessionUsage(new Exception("trace"));
}
}
+ @Override
public void endCall() {
concurrentCall.decrementAndGet();
}
@@ -1032,6 +1118,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
// XAResource implementation
// --------------------------------------------------------------------
+ @Override
public void commit(final Xid xid, final boolean onePhase) throws XAException {
if (ActiveMQClientLogger.LOGGER.isTraceEnabled()) {
ActiveMQClientLogger.LOGGER.trace("call commit(xid=" + convert(xid));
@@ -1073,6 +1160,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public void end(final Xid xid, final int flags) throws XAException {
if (ActiveMQClientLogger.LOGGER.isTraceEnabled()) {
ActiveMQClientLogger.LOGGER.trace("Calling end:: " + convert(xid) + ", flags=" + convertTXFlag(flags));
@@ -1118,6 +1206,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public void forget(final Xid xid) throws XAException {
checkXA();
startCall();
@@ -1138,6 +1227,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public int getTransactionTimeout() throws XAException {
checkXA();
@@ -1152,6 +1242,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public boolean setTransactionTimeout(final int seconds) throws XAException {
checkXA();
@@ -1166,6 +1257,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public boolean isSameRM(final XAResource xares) throws XAException {
checkXA();
@@ -1203,6 +1295,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
return null;
}
+ @Override
public int prepare(final Xid xid) throws XAException {
checkXA();
if (ActiveMQClientLogger.LOGGER.isTraceEnabled()) {
@@ -1269,6 +1362,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public Xid[] recover(final int flags) throws XAException {
checkXA();
@@ -1287,6 +1381,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
return new Xid[0];
}
+ @Override
public void rollback(final Xid xid) throws XAException {
checkXA();
@@ -1341,6 +1436,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public void start(final Xid xid, final int flags) throws XAException {
if (ActiveMQClientLogger.LOGGER.isTraceEnabled()) {
ActiveMQClientLogger.LOGGER.trace("Calling start:: " + convert(xid) + " clientXID=" + xid + " flags = " + convertTXFlag(flags));
@@ -1388,6 +1484,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
// FailureListener implementation --------------------------------------------
+ @Override
public void connectionFailed(final ActiveMQException me, boolean failedOver) {
try {
cleanUp(false);
@@ -1397,6 +1494,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
}
}
+ @Override
public void connectionFailed(final ActiveMQException me, boolean failedOver, String scaleDownTargetNodeID) {
connectionFailed(me, failedOver);
}
@@ -1404,10 +1502,12 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
// Public
// ----------------------------------------------------------------------------
+ @Override
public void setForceNotSameRM(final boolean force) {
forceNotSameRM = force;
}
+ @Override
public RemotingConnection getConnection() {
return sessionContext.getRemotingConnection();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java
index ae711bf..0cb38f6 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java
@@ -47,6 +47,7 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
/**
*
*/
+ @Override
public void discardUnusedPackets() {
bufferDelegate.discardUnusedPackets();
}
@@ -54,22 +55,27 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
/**
* Add a buff to the List, or save it to the OutputStream if set
*/
+ @Override
public void addPacket(byte[] chunk, int flowControlSize, boolean isContinues) {
bufferDelegate.addPacket(chunk, flowControlSize, isContinues);
}
+ @Override
public synchronized void cancel() {
bufferDelegate.cancel();
}
+ @Override
public synchronized void close() {
bufferDelegate.cancel();
}
+ @Override
public void setOutputStream(final OutputStream output) throws ActiveMQException {
bufferDelegate.setOutputStream(new InflaterWriter(output));
}
+ @Override
public synchronized void saveBuffer(final OutputStream output) throws ActiveMQException {
setOutputStream(output);
waitCompletion(0);
@@ -78,6 +84,7 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
/**
* @param timeWait Milliseconds to Wait. 0 means forever
*/
+ @Override
public synchronized boolean waitCompletion(final long timeWait) throws ActiveMQException {
return bufferDelegate.waitCompletion(timeWait);
}
@@ -108,6 +115,7 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
throw new IllegalStateException("Position not supported over compressed large messages");
}
+ @Override
public byte readByte() {
try {
return getStream().readByte();
@@ -196,82 +204,102 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public int readerIndex() {
return 0;
}
+ @Override
public void readerIndex(final int readerIndex) {
// TODO
}
+ @Override
public int writerIndex() {
// TODO
return 0;
}
+ @Override
public long getSize() {
return this.bufferDelegate.getSize();
}
+ @Override
public void writerIndex(final int writerIndex) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void setIndex(final int readerIndex, final int writerIndex) {
positioningNotSupported();
}
+ @Override
public void clear() {
}
+ @Override
public boolean readable() {
return true;
}
+ @Override
public boolean writable() {
return false;
}
+ @Override
public int readableBytes() {
return 1;
}
+ @Override
public int writableBytes() {
return 0;
}
+ @Override
public void markReaderIndex() {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void resetReaderIndex() {
// TODO: reset positioning if possible
}
+ @Override
public void markWriterIndex() {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void resetWriterIndex() {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void discardReadBytes() {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public short getUnsignedByte(final int index) {
return (short) (getByte(index) & 0xFF);
}
+ @Override
public int getUnsignedShort(final int index) {
return getShort(index) & 0xFFFF;
}
+ @Override
public long getUnsignedInt(final int index) {
return getInt(index) & 0xFFFFFFFFL;
}
+ @Override
public void getBytes(int index, final byte[] dst) {
// TODO: optimize this by using System.arraycopy
for (int i = 0; i < dst.length; i++) {
@@ -279,10 +307,12 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
}
}
+ @Override
public void getBytes(final int index, final ActiveMQBuffer dst) {
getBytes(index, dst, dst.writableBytes());
}
+ @Override
public void getBytes(final int index, final ActiveMQBuffer dst, final int length) {
if (length > dst.writableBytes()) {
throw new IndexOutOfBoundsException();
@@ -291,18 +321,22 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
dst.writerIndex(dst.writerIndex() + length);
}
+ @Override
public void setBytes(final int index, final byte[] src) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void setBytes(final int index, final ActiveMQBuffer src) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void setBytes(final int index, final ActiveMQBuffer src, final int length) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public int readUnsignedByte() {
try {
return getStream().readUnsignedByte();
@@ -312,6 +346,7 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
}
}
+ @Override
public short readShort() {
try {
return getStream().readShort();
@@ -321,6 +356,7 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
}
}
+ @Override
public int readUnsignedShort() {
try {
return getStream().readUnsignedShort();
@@ -330,6 +366,7 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
}
}
+ @Override
public int readInt() {
try {
return getStream().readInt();
@@ -339,10 +376,12 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
}
}
+ @Override
public long readUnsignedInt() {
return readInt() & 0xFFFFFFFFL;
}
+ @Override
public long readLong() {
try {
return getStream().readLong();
@@ -352,6 +391,7 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
}
}
+ @Override
public void readBytes(final byte[] dst, final int dstIndex, final int length) {
try {
int nReadBytes = getStream().read(dst, dstIndex, length);
@@ -364,14 +404,17 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
}
}
+ @Override
public void readBytes(final byte[] dst) {
readBytes(dst, 0, dst.length);
}
+ @Override
public void readBytes(final ActiveMQBuffer dst) {
readBytes(dst, dst.writableBytes());
}
+ @Override
public void readBytes(final ActiveMQBuffer dst, final int length) {
if (length > dst.writableBytes()) {
throw new IndexOutOfBoundsException();
@@ -380,18 +423,21 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
dst.writerIndex(dst.writerIndex() + length);
}
+ @Override
public void readBytes(final ActiveMQBuffer dst, final int dstIndex, final int length) {
byte[] destBytes = new byte[length];
readBytes(destBytes);
dst.setBytes(dstIndex, destBytes);
}
+ @Override
public void readBytes(final ByteBuffer dst) {
byte[] bytesToGet = new byte[dst.remaining()];
readBytes(bytesToGet);
dst.put(bytesToGet);
}
+ @Override
public int skipBytes(final int length) {
try {
@@ -425,38 +471,47 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
}
+ @Override
public void writeByte(final byte value) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void writeShort(final short value) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void writeInt(final int value) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void writeLong(final long value) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void writeBytes(final byte[] src, final int srcIndex, final int length) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void writeBytes(final byte[] src) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void writeBytes(final ActiveMQBuffer src, final int length) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void writeBytes(final ByteBuffer src) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public ByteBuffer toByteBuffer() {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
@@ -475,18 +530,22 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
return (char) readShort();
}
+ @Override
public char getChar(final int index) {
return (char) getShort(index);
}
+ @Override
public double getDouble(final int index) {
return Double.longBitsToDouble(getLong(index));
}
+ @Override
public float getFloat(final int index) {
return Float.intBitsToFloat(getInt(index));
}
+ @Override
public ActiveMQBuffer readBytes(final int length) {
byte[] bytesToGet = new byte[length];
readBytes(bytesToGet);
@@ -604,48 +663,59 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public ActiveMQBuffer copy() {
throw new UnsupportedOperationException();
}
+ @Override
public ActiveMQBuffer slice(final int index, final int length) {
throw new UnsupportedOperationException();
}
// Inner classes -------------------------------------------------
+ @Override
public ByteBuf byteBuf() {
return null;
}
+ @Override
public ActiveMQBuffer copy(final int index, final int length) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public ActiveMQBuffer duplicate() {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public ActiveMQBuffer readSlice(final int length) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void setChar(final int index, final char value) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void setDouble(final int index, final double value) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void setFloat(final int index, final float value) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public ActiveMQBuffer slice() {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
public void writeBytes(final ActiveMQBuffer src, final int srcIndex, final int length) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/DelegatingSession.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/DelegatingSession.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/DelegatingSession.java
index 4d72dc9..95bb26f 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/DelegatingSession.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/DelegatingSession.java
@@ -90,10 +90,12 @@ public class DelegatingSession implements ClientSessionInternal {
}
}
+ @Override
public boolean isClosing() {
return session.isClosing();
}
+ @Override
public void acknowledge(final ClientConsumer consumer, final Message message) throws ActiveMQException {
session.acknowledge(consumer, message);
}
@@ -103,34 +105,42 @@ public class DelegatingSession implements ClientSessionInternal {
session.addLifeCycleListener(lifeCycleListener);
}
+ @Override
public void individualAcknowledge(final ClientConsumer consumer, final Message message) throws ActiveMQException {
session.individualAcknowledge(consumer, message);
}
+ @Override
public void addConsumer(final ClientConsumerInternal consumer) {
session.addConsumer(consumer);
}
+ @Override
public void addFailureListener(final SessionFailureListener listener) {
session.addFailureListener(listener);
}
+ @Override
public void addFailoverListener(FailoverEventListener listener) {
session.addFailoverListener(listener);
}
+ @Override
public void addProducer(final ClientProducerInternal producer) {
session.addProducer(producer);
}
+ @Override
public AddressQuery addressQuery(final SimpleString address) throws ActiveMQException {
return session.addressQuery(address);
}
+ @Override
public void cleanUp(boolean failingOver) throws ActiveMQException {
session.cleanUp(failingOver);
}
+ @Override
public void close() throws ActiveMQException {
closed = true;
@@ -141,22 +151,27 @@ public class DelegatingSession implements ClientSessionInternal {
session.close();
}
+ @Override
public void markRollbackOnly() {
session.markRollbackOnly();
}
+ @Override
public void commit() throws ActiveMQException {
session.commit();
}
+ @Override
public void commit(final Xid xid, final boolean onePhase) throws XAException {
session.commit(xid, onePhase);
}
+ @Override
public ClientMessage createMessage(final boolean durable) {
return session.createMessage(durable);
}
+ @Override
public ClientMessage createMessage(final byte type,
final boolean durable,
final long expiration,
@@ -165,16 +180,19 @@ public class DelegatingSession implements ClientSessionInternal {
return session.createMessage(type, durable, expiration, timestamp, priority);
}
+ @Override
public ClientMessage createMessage(final byte type, final boolean durable) {
return session.createMessage(type, durable);
}
+ @Override
public ClientConsumer createConsumer(final SimpleString queueName,
final SimpleString filterString,
final boolean browseOnly) throws ActiveMQException {
return session.createConsumer(queueName, filterString, browseOnly);
}
+ @Override
public ClientConsumer createConsumer(final SimpleString queueName,
final SimpleString filterString,
final int windowSize,
@@ -183,21 +201,25 @@ public class DelegatingSession implements ClientSessionInternal {
return session.createConsumer(queueName, filterString, windowSize, maxRate, browseOnly);
}
+ @Override
public ClientConsumer createConsumer(final SimpleString queueName,
final SimpleString filterString) throws ActiveMQException {
return session.createConsumer(queueName, filterString);
}
+ @Override
public ClientConsumer createConsumer(final SimpleString queueName) throws ActiveMQException {
return session.createConsumer(queueName);
}
+ @Override
public ClientConsumer createConsumer(final String queueName,
final String filterString,
final boolean browseOnly) throws ActiveMQException {
return session.createConsumer(queueName, filterString, browseOnly);
}
+ @Override
public ClientConsumer createConsumer(final String queueName,
final String filterString,
final int windowSize,
@@ -206,47 +228,58 @@ public class DelegatingSession implements ClientSessionInternal {
return session.createConsumer(queueName, filterString, windowSize, maxRate, browseOnly);
}
+ @Override
public ClientConsumer createConsumer(final String queueName, final String filterString) throws ActiveMQException {
return session.createConsumer(queueName, filterString);
}
+ @Override
public ClientConsumer createConsumer(final String queueName) throws ActiveMQException {
return session.createConsumer(queueName);
}
+ @Override
public ClientConsumer createConsumer(final SimpleString queueName,
final boolean browseOnly) throws ActiveMQException {
return session.createConsumer(queueName, browseOnly);
}
+ @Override
public ClientConsumer createConsumer(final String queueName, final boolean browseOnly) throws ActiveMQException {
return session.createConsumer(queueName, browseOnly);
}
+ @Override
public ClientProducer createProducer() throws ActiveMQException {
return session.createProducer();
}
+ @Override
public ClientProducer createProducer(final SimpleString address, final int rate) throws ActiveMQException {
return session.createProducer(address, rate);
}
+ @Override
public ClientProducer createProducer(final SimpleString address) throws ActiveMQException {
return session.createProducer(address);
}
+ @Override
public ClientProducer createProducer(final String address) throws ActiveMQException {
return session.createProducer(address);
}
+ @Override
public void createQueue(final String address, final String queueName) throws ActiveMQException {
session.createQueue(address, queueName);
}
+ @Override
public void createQueue(final SimpleString address, final SimpleString queueName) throws ActiveMQException {
session.createQueue(address, queueName);
}
+ @Override
public void createQueue(final SimpleString address,
final SimpleString queueName,
final boolean durable) throws ActiveMQException {
@@ -268,6 +301,7 @@ public class DelegatingSession implements ClientSessionInternal {
session.createSharedQueue(address, queueName, filter, durable);
}
+ @Override
public void createQueue(final SimpleString address,
final SimpleString queueName,
final SimpleString filterString,
@@ -275,12 +309,14 @@ public class DelegatingSession implements ClientSessionInternal {
session.createQueue(address, queueName, filterString, durable);
}
+ @Override
public void createQueue(final String address,
final String queueName,
final boolean durable) throws ActiveMQException {
session.createQueue(address, queueName, durable);
}
+ @Override
public void createQueue(final String address,
final String queueName,
final String filterString,
@@ -288,74 +324,91 @@ public class DelegatingSession implements ClientSessionInternal {
session.createQueue(address, queueName, filterString, durable);
}
+ @Override
public void createTemporaryQueue(final SimpleString address,
final SimpleString queueName,
final SimpleString filter) throws ActiveMQException {
session.createTemporaryQueue(address, queueName, filter);
}
+ @Override
public void createTemporaryQueue(final SimpleString address, final SimpleString queueName) throws ActiveMQException {
session.createTemporaryQueue(address, queueName);
}
+ @Override
public void createTemporaryQueue(final String address,
final String queueName,
final String filter) throws ActiveMQException {
session.createTemporaryQueue(address, queueName, filter);
}
+ @Override
public void createTemporaryQueue(final String address, final String queueName) throws ActiveMQException {
session.createTemporaryQueue(address, queueName);
}
+ @Override
public void deleteQueue(final SimpleString queueName) throws ActiveMQException {
session.deleteQueue(queueName);
}
+ @Override
public void deleteQueue(final String queueName) throws ActiveMQException {
session.deleteQueue(queueName);
}
+ @Override
public void end(final Xid xid, final int flags) throws XAException {
session.end(xid, flags);
}
+ @Override
public void expire(final ClientConsumer consumer, final Message message) throws ActiveMQException {
session.expire(consumer, message);
}
+ @Override
public void forget(final Xid xid) throws XAException {
session.forget(xid);
}
+ @Override
public RemotingConnection getConnection() {
return session.getConnection();
}
+ @Override
public int getMinLargeMessageSize() {
return session.getMinLargeMessageSize();
}
+ @Override
public String getName() {
return session.getName();
}
+ @Override
public int getTransactionTimeout() throws XAException {
return session.getTransactionTimeout();
}
+ @Override
public int getVersion() {
return session.getVersion();
}
+ @Override
public XAResource getXAResource() {
return session.getXAResource();
}
+ @Override
public void preHandleFailover(RemotingConnection connection) {
session.preHandleFailover(connection);
}
+ @Override
public void handleFailover(final RemotingConnection backupConnection, ActiveMQException cause) {
session.handleFailover(backupConnection, cause);
}
@@ -385,152 +438,189 @@ public class DelegatingSession implements ClientSessionInternal {
session.handleConsumerDisconnect(consumerContext);
}
+ @Override
public boolean isAutoCommitAcks() {
return session.isAutoCommitAcks();
}
+ @Override
public boolean isAutoCommitSends() {
return session.isAutoCommitSends();
}
+ @Override
public boolean isBlockOnAcknowledge() {
return session.isBlockOnAcknowledge();
}
+ @Override
public boolean isCacheLargeMessageClient() {
return session.isCacheLargeMessageClient();
}
+ @Override
public boolean isClosed() {
return session.isClosed();
}
+ @Override
public boolean isSameRM(final XAResource xares) throws XAException {
return session.isSameRM(xares);
}
+ @Override
public boolean isXA() {
return session.isXA();
}
+ @Override
public int prepare(final Xid xid) throws XAException {
return session.prepare(xid);
}
+ @Override
public QueueQuery queueQuery(final SimpleString queueName) throws ActiveMQException {
return session.queueQuery(queueName);
}
+ @Override
public Xid[] recover(final int flag) throws XAException {
return session.recover(flag);
}
+ @Override
public void removeConsumer(final ClientConsumerInternal consumer) throws ActiveMQException {
session.removeConsumer(consumer);
}
+ @Override
public boolean removeFailureListener(final SessionFailureListener listener) {
return session.removeFailureListener(listener);
}
+ @Override
public boolean removeFailoverListener(FailoverEventListener listener) {
return session.removeFailoverListener(listener);
}
+ @Override
public void removeProducer(final ClientProducerInternal producer) {
session.removeProducer(producer);
}
+ @Override
public void rollback() throws ActiveMQException {
session.rollback();
}
+ @Override
public boolean isRollbackOnly() {
return session.isRollbackOnly();
}
+ @Override
public void rollback(final boolean considerLastMessageAsDelivered) throws ActiveMQException {
session.rollback(considerLastMessageAsDelivered);
}
+ @Override
public void rollback(final Xid xid) throws XAException {
session.rollback(xid);
}
+ @Override
public DelegatingSession setSendAcknowledgementHandler(final SendAcknowledgementHandler handler) {
session.setSendAcknowledgementHandler(handler);
return this;
}
+ @Override
public boolean setTransactionTimeout(final int seconds) throws XAException {
return session.setTransactionTimeout(seconds);
}
+ @Override
public void resetIfNeeded() throws ActiveMQException {
session.resetIfNeeded();
}
+ @Override
public DelegatingSession start() throws ActiveMQException {
session.start();
return this;
}
+ @Override
public void start(final Xid xid, final int flags) throws XAException {
session.start(xid, flags);
}
+ @Override
public void stop() throws ActiveMQException {
session.stop();
}
+ @Override
public ClientSessionFactory getSessionFactory() {
return session.getSessionFactory();
}
+ @Override
public void setForceNotSameRM(final boolean force) {
session.setForceNotSameRM(force);
}
+ @Override
public void workDone() {
session.workDone();
}
+ @Override
public void sendProducerCreditsMessage(final int credits, final SimpleString address) {
session.sendProducerCreditsMessage(credits, address);
}
+ @Override
public ClientProducerCredits getCredits(final SimpleString address, final boolean anon) {
return session.getCredits(address, anon);
}
+ @Override
public void returnCredits(final SimpleString address) {
session.returnCredits(address);
}
+ @Override
public void handleReceiveProducerCredits(final SimpleString address, final int credits) {
session.handleReceiveProducerCredits(address, credits);
}
+ @Override
public void handleReceiveProducerFailCredits(final SimpleString address, final int credits) {
session.handleReceiveProducerFailCredits(address, credits);
}
+ @Override
public ClientProducerCreditManager getProducerCreditManager() {
return session.getProducerCreditManager();
}
+ @Override
public void setAddress(Message message, SimpleString address) {
session.setAddress(message, address);
}
+ @Override
public void setPacketSize(int packetSize) {
session.setPacketSize(packetSize);
}
+ @Override
public void addMetaData(String key, String data) throws ActiveMQException {
session.addMetaData(key, data);
}
+ @Override
public boolean isCompressLargeMessages() {
return session.isCompressLargeMessages();
}
@@ -546,10 +636,12 @@ public class DelegatingSession implements ClientSessionInternal {
}
+ @Override
public void startCall() {
session.startCall();
}
+ @Override
public void endCall() {
session.endCall();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java
index ad79e82..fb5b687 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java
@@ -134,6 +134,7 @@ public class LargeMessageControllerImpl implements LargeMessageController {
this.local = local;
}
+ @Override
public void discardUnusedPackets() {
if (outStream == null) {
if (local)
@@ -150,6 +151,7 @@ public class LargeMessageControllerImpl implements LargeMessageController {
* TODO: move this to ConsumerContext as large message is a protocol specific thing
* Add a buff to the List, or save it to the OutputStream if set
*/
+ @Override
public void addPacket(byte[] chunk, int flowControlSize, boolean isContinues) {
int flowControlCredit = 0;
@@ -206,6 +208,7 @@ public class LargeMessageControllerImpl implements LargeMessageController {
}
}
+ @Override
public void cancel() {
this.handledException = ActiveMQClientMessageBundle.BUNDLE.largeMessageInterrupted();
@@ -232,12 +235,14 @@ public class LargeMessageControllerImpl implements LargeMessageController {
}
}
+ @Override
public synchronized void close() {
if (fileCache != null) {
fileCache.close();
}
}
+ @Override
public void setOutputStream(final OutputStream output) throws ActiveMQException {
int totalFlowControl = 0;
@@ -268,6 +273,7 @@ public class LargeMessageControllerImpl implements LargeMessageController {
}
}
+ @Override
public synchronized void saveBuffer(final OutputStream output) throws ActiveMQException {
if (streamClosed) {
throw ActiveMQClientMessageBundle.BUNDLE.largeMessageLostSession();
@@ -280,6 +286,7 @@ public class LargeMessageControllerImpl implements LargeMessageController {
* @param timeWait Milliseconds to Wait. 0 means forever
* @throws ActiveMQException
*/
+ @Override
public synchronized boolean waitCompletion(final long timeWait) throws ActiveMQException {
if (outStream == null) {
// There is no stream.. it will never achieve the end of streaming
@@ -344,6 +351,7 @@ public class LargeMessageControllerImpl implements LargeMessageController {
return -1;
}
+ @Override
public byte readByte() {
return getByte(readerIndex++);
}
@@ -425,6 +433,7 @@ public class LargeMessageControllerImpl implements LargeMessageController {
return out.write(ByteBuffer.wrap(bytesToGet));
}
+ @Override
public int getInt(final int index) {
return (getByte(index) & 0xff) << 24 | (getByte(index + 1) & 0xff) << 16 |
(getByte(index + 2) & 0xff) << 8 |
@@ -515,10 +524,12 @@ public class LargeMessageControllerImpl implements LargeMessageController {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public int readerIndex() {
return (int) readerIndex;
}
+ @Override
public void readerIndex(final int readerIndex) {
try {
checkForPacket(readerIndex);
@@ -530,18 +541,22 @@ public class LargeMessageControllerImpl implements LargeMessageController {
this.readerIndex = readerIndex;
}
+ @Override
public int writerIndex() {
return (int) totalSize;
}
+ @Override
public long getSize() {
return totalSize;
}
+ @Override
public void writerIndex(final int writerIndex) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void setIndex(final int readerIndex, final int writerIndex) {
try {
checkForPacket(readerIndex);
@@ -553,17 +568,21 @@ public class LargeMessageControllerImpl implements LargeMessageController {
this.readerIndex = readerIndex;
}
+ @Override
public void clear() {
}
+ @Override
public boolean readable() {
return true;
}
+ @Override
public boolean writable() {
return false;
}
+ @Override
public int readableBytes() {
long readableBytes = totalSize - readerIndex;
@@ -575,14 +594,17 @@ public class LargeMessageControllerImpl implements LargeMessageController {
}
}
+ @Override
public int writableBytes() {
return 0;
}
+ @Override
public void markReaderIndex() {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void resetReaderIndex() {
try {
checkForPacket(0);
@@ -593,22 +615,27 @@ public class LargeMessageControllerImpl implements LargeMessageController {
}
}
+ @Override
public void markWriterIndex() {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void resetWriterIndex() {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void discardReadBytes() {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public short getUnsignedByte(final int index) {
return (short) (getByte(index) & 0xFF);
}
+ @Override
public int getUnsignedShort(final int index) {
return getShort(index) & 0xFFFF;
}
@@ -621,10 +648,12 @@ public class LargeMessageControllerImpl implements LargeMessageController {
return value;
}
+ @Override
public long getUnsignedInt(final int index) {
return getInt(index) & 0xFFFFFFFFL;
}
+ @Override
public void getBytes(int index, final byte[] dst) {
// TODO: optimize this by using System.arraycopy
for (int i = 0; i < dst.length; i++) {
@@ -639,10 +668,12 @@ public class LargeMessageControllerImpl implements LargeMessageController {
}
}
+ @Override
public void getBytes(final int index, final ActiveMQBuffer dst) {
getBytes(index, dst, dst.writableBytes());
}
+ @Override
public void getBytes(final int index, final ActiveMQBuffer dst, final int length) {
if (length > dst.writableBytes()) {
throw new IndexOutOfBoundsException();
@@ -651,14 +682,17 @@ public class LargeMessageControllerImpl implements LargeMessageController {
dst.writerIndex(dst.writerIndex() + length);
}
+ @Override
public void setBytes(final int index, final byte[] src) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void setBytes(final int index, final ActiveMQBuffer src) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void setBytes(final int index, final ActiveMQBuffer src, final int length) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
@@ -667,16 +701,19 @@ public class LargeMessageControllerImpl implements LargeMessageController {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public int readUnsignedByte() {
return (short) (readByte() & 0xFF);
}
+ @Override
public short readShort() {
short v = getShort(readerIndex);
readerIndex += 2;
return v;
}
+ @Override
public int readUnsignedShort() {
return readShort() & 0xFFFF;
}
@@ -695,6 +732,7 @@ public class LargeMessageControllerImpl implements LargeMessageController {
return v;
}
+ @Override
public int readInt() {
int v = getInt(readerIndex);
readerIndex += 4;
@@ -706,29 +744,35 @@ public class LargeMessageControllerImpl implements LargeMessageController {
return v;
}
+ @Override
public long readUnsignedInt() {
return readInt() & 0xFFFFFFFFL;
}
+ @Override
public long readLong() {
long v = getLong(readerIndex);
readerIndex += 8;
return v;
}
+ @Override
public void readBytes(final byte[] dst, final int dstIndex, final int length) {
getBytes(readerIndex, dst, dstIndex, length);
readerIndex += length;
}
+ @Override
public void readBytes(final byte[] dst) {
readBytes(dst, 0, dst.length);
}
+ @Override
public void readBytes(final ActiveMQBuffer dst) {
readBytes(dst, dst.writableBytes());
}
+ @Override
public void readBytes(final ActiveMQBuffer dst, final int length) {
if (length > dst.writableBytes()) {
throw new IndexOutOfBoundsException();
@@ -737,11 +781,13 @@ public class LargeMessageControllerImpl implements LargeMessageController {
dst.writerIndex(dst.writerIndex() + length);
}
+ @Override
public void readBytes(final ActiveMQBuffer dst, final int dstIndex, final int length) {
getBytes(readerIndex, dst, dstIndex, length);
readerIndex += length;
}
+ @Override
public void readBytes(final ByteBuffer dst) {
int length = dst.remaining();
getBytes(readerIndex, dst);
@@ -759,6 +805,7 @@ public class LargeMessageControllerImpl implements LargeMessageController {
readerIndex += length;
}
+ @Override
public int skipBytes(final int length) {
long newReaderIndex = readerIndex + length;
@@ -767,10 +814,12 @@ public class LargeMessageControllerImpl implements LargeMessageController {
return length;
}
+ @Override
public void writeByte(final byte value) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void writeShort(final short value) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
@@ -779,18 +828,22 @@ public class LargeMessageControllerImpl implements LargeMessageController {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void writeInt(final int value) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void writeLong(final long value) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void writeBytes(final byte[] src, final int srcIndex, final int length) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void writeBytes(final byte[] src) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
@@ -799,10 +852,12 @@ public class LargeMessageControllerImpl implements LargeMessageController {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void writeBytes(final ActiveMQBuffer src, final int length) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void writeBytes(final ByteBuffer src) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
@@ -819,6 +874,7 @@ public class LargeMessageControllerImpl implements LargeMessageController {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public ByteBuffer toByteBuffer() {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
@@ -849,18 +905,22 @@ public class LargeMessageControllerImpl implements LargeMessageController {
return (char) readShort();
}
+ @Override
public char getChar(final int index) {
return (char) getShort(index);
}
+ @Override
public double getDouble(final int index) {
return Double.longBitsToDouble(getLong(index));
}
+ @Override
public float getFloat(final int index) {
return Float.intBitsToFloat(getInt(index));
}
+ @Override
public ActiveMQBuffer readBytes(final int length) {
byte[] bytesToGet = new byte[length];
getBytes(readerIndex, bytesToGet);
@@ -979,10 +1039,12 @@ public class LargeMessageControllerImpl implements LargeMessageController {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public ActiveMQBuffer copy() {
throw new UnsupportedOperationException();
}
+ @Override
public ActiveMQBuffer slice(final int index, final int length) {
throw new UnsupportedOperationException();
}
@@ -1198,38 +1260,47 @@ public class LargeMessageControllerImpl implements LargeMessageController {
return ByteUtil.readLine(this);
}
+ @Override
public ByteBuf byteBuf() {
return null;
}
+ @Override
public ActiveMQBuffer copy(final int index, final int length) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public ActiveMQBuffer duplicate() {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public ActiveMQBuffer readSlice(final int length) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void setChar(final int index, final char value) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void setDouble(final int index, final double value) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void setFloat(final int index, final float value) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public ActiveMQBuffer slice() {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
public void writeBytes(final ActiveMQBuffer src, final int srcIndex, final int length) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/QueueQueryImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/QueueQueryImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/QueueQueryImpl.java
index 27607bc..40ea86a 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/QueueQueryImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/QueueQueryImpl.java
@@ -70,38 +70,47 @@ public class QueueQueryImpl implements ClientSession.QueueQuery {
this.autoCreateJmsQueues = autoCreateJmsQueues;
}
+ @Override
public SimpleString getName() {
return name;
}
+ @Override
public SimpleString getAddress() {
return address;
}
+ @Override
public int getConsumerCount() {
return consumerCount;
}
+ @Override
public SimpleString getFilterString() {
return filterString;
}
+ @Override
public long getMessageCount() {
return messageCount;
}
+ @Override
public boolean isDurable() {
return durable;
}
+ @Override
public boolean isAutoCreateJmsQueues() {
return autoCreateJmsQueues;
}
+ @Override
public boolean isTemporary() {
return temporary;
}
+ @Override
public boolean isExists() {
return exists;
}