You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kg...@apache.org on 2011/09/19 17:13:38 UTC
svn commit: r1172657 [8/21] - in /qpid/branches/qpid-3346/qpid: ./ cpp/
cpp/bindings/ cpp/bindings/qmf2/examples/cpp/ cpp/bindings/qpid/dotnet/
cpp/bindings/qpid/dotnet/examples/csharp.direct.receiver/Properties/
cpp/bindings/qpid/dotnet/examples/cshar...
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java Mon Sep 19 15:13:18 2011
@@ -30,15 +30,12 @@ import org.apache.qpid.server.queue.AMQQ
import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.message.InboundMessage;
-import org.apache.qpid.server.binding.BindingFactory;
import org.apache.qpid.server.binding.Binding;
import org.apache.qpid.server.configuration.ExchangeConfig;
import javax.management.JMException;
import java.util.ArrayList;
-import java.util.List;
import java.util.Collection;
-import java.util.concurrent.CopyOnWriteArrayList;
public interface Exchange extends ExchangeReferrer, ExchangeConfig
{
@@ -67,7 +64,12 @@ public interface Exchange extends Exchan
void close() throws AMQException;
-
+ /**
+ * Returns a list of queues to which to route this message. If there are
+ * no queues the empty list must be returned.
+ *
+ * @return list of queues to which to route the message.
+ */
ArrayList<? extends BaseQueue> route(InboundMessage message);
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeadersParser.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeadersParser.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeadersParser.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeadersParser.java Mon Sep 19 15:13:18 2011
@@ -274,132 +274,6 @@ public class HeadersParser
}
- public static void main(String[] args) throws AMQFrameDecodingException
- {
-
- FieldTable bindingTable = new FieldTable();
-
- bindingTable.setString(new AMQShortString("x-match"),"all");
- bindingTable.setInteger("a",1);
- bindingTable.setVoid(new AMQShortString("b"));
- bindingTable.setString("c","");
- bindingTable.setInteger("d",4);
- bindingTable.setInteger("e",1);
-
-
-
- FieldTable bindingTable2 = new FieldTable();
- bindingTable2.setString(new AMQShortString("x-match"),"all");
- bindingTable2.setInteger("a",1);
- bindingTable2.setVoid(new AMQShortString("b"));
- bindingTable2.setString("c","");
- bindingTable2.setInteger("d",4);
- bindingTable2.setInteger("e",1);
- bindingTable2.setInteger("f",1);
-
-
- FieldTable table = new FieldTable();
- table.setInteger("a",1);
- table.setInteger("b",2);
- table.setString("c","");
- table.setInteger("d",4);
- table.setInteger("e",1);
- table.setInteger("f",1);
- table.setInteger("h",1);
- table.setInteger("i",1);
- table.setInteger("j",1);
- table.setInteger("k",1);
- table.setInteger("l",1);
-
- org.apache.mina.common.ByteBuffer buffer = org.apache.mina.common.ByteBuffer.allocate( (int) table.getEncodedSize());
- EncodingUtils.writeFieldTableBytes(buffer, table);
- buffer.flip();
-
- FieldTable table2 = EncodingUtils.readFieldTable(buffer);
-
-
-
- FieldTable bindingTable3 = new FieldTable();
- bindingTable3.setString(new AMQShortString("x-match"),"any");
- bindingTable3.setInteger("a",1);
- bindingTable3.setInteger("b",3);
-
-
- FieldTable bindingTable4 = new FieldTable();
- bindingTable4.setString(new AMQShortString("x-match"),"any");
- bindingTable4.setVoid(new AMQShortString("a"));
-
-
- FieldTable bindingTable5 = new FieldTable();
- bindingTable5.setString(new AMQShortString("x-match"),"all");
- bindingTable5.setString(new AMQShortString("h"),"hello");
-
- for(int i = 0; i < 100; i++)
- {
- printMatches(new FieldTable[] {bindingTable5} , table2);
- }
-
-
-
- }
-
-
-
- private static void printMatches(final FieldTable[] bindingKeys, final FieldTable routingKey)
- {
- HeadersMatcherDFAState sm = null;
- Map<HeaderMatcherResult, String> resultMap = new HashMap<HeaderMatcherResult, String>();
-
- HeadersParser parser = new HeadersParser();
-
- for(int i = 0; i < bindingKeys.length; i++)
- {
- HeaderMatcherResult r = new HeaderMatcherResult();
- resultMap.put(r, bindingKeys[i].toString());
-
-
- if(i==0)
- {
- sm = parser.createStateMachine(bindingKeys[i], r);
- }
- else
- {
- sm = sm.mergeStateMachines(parser.createStateMachine(bindingKeys[i], r));
- }
- }
-
- Collection<HeaderMatcherResult> results = null;
- long beforeTime = System.currentTimeMillis();
- for(int i = 0; i < 1000000; i++)
- {
- routingKey.size();
-
- assert sm != null;
- results = sm.match(routingKey);
-
- }
- long elapsed = System.currentTimeMillis() - beforeTime;
- System.out.println("1000000 Iterations took: " + elapsed);
- Collection<String> resultStrings = new ArrayList<String>();
-
- assert results != null;
- for(HeaderMatcherResult result : results)
- {
- resultStrings.add(resultMap.get(result));
- }
-
- final ArrayList<String> nonMatches = new ArrayList<String>();
- for(FieldTable key : bindingKeys)
- {
- nonMatches.add(key.toString());
- }
- nonMatches.removeAll(resultStrings);
- System.out.println("\""+routingKey+"\" matched with " + resultStrings + " DID NOT MATCH with " + nonMatches);
-
-
- }
-
-
public final static class KeyValuePair
{
public final HeaderKey _key;
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java Mon Sep 19 15:13:18 2011
@@ -38,6 +38,7 @@ import org.apache.qpid.server.queue.Base
import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.server.subscription.SubscriptionFactoryImpl;
import org.apache.qpid.server.subscription.Subscription_0_10;
import org.apache.qpid.server.transport.ServerSession;
import org.apache.qpid.server.txn.AutoCommitTransaction;
@@ -696,7 +697,7 @@ public class Bridge implements BridgeCon
//TODO Handle the passing of non-null Filters and Arguments here
- Subscription_0_10 sub = new Subscription_0_10((ServerSession)session,
+ Subscription_0_10 sub = SubscriptionFactoryImpl.INSTANCE.createSubscription((ServerSession)session,
_destination,
MessageAcceptMode.NONE,
MessageAcquireMode.PRE_ACQUIRED,
@@ -768,7 +769,7 @@ public class Bridge implements BridgeCon
//TODO Handle the passing of non-null Filters and Arguments here
- Subscription_0_10 sub = new Subscription_0_10((ServerSession)session,
+ Subscription_0_10 sub = SubscriptionFactoryImpl.INSTANCE.createSubscription((ServerSession)session,
_destination,
MessageAcceptMode.NONE,
MessageAcquireMode.PRE_ACQUIRED,
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java Mon Sep 19 15:13:18 2011
@@ -258,7 +258,6 @@ public class BrokerLink implements LinkC
_remoteFederationTag = UUID.fromString(_transport+":"+_host+":"+_port).toString();
}
_qpidConnection.setSessionFactory(new SessionFactory());
- _qpidConnection.setAuthorizationID(_username == null ? "" : _username);
updateState(State.ESTABLISHING, State.OPERATIONAL);
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java Mon Sep 19 15:13:18 2011
@@ -162,14 +162,7 @@ public class BasicGetMethodHandler imple
}
else
{
- sub = new GetNoAckSubscription(channel,
- session,
- null,
- null,
- false,
- singleMessageCredit,
- getDeliveryMethod,
- getRecordMethod);
+ sub = SubscriptionFactoryImpl.INSTANCE.createBasicGetNoAckSubscription(channel, session, null, null, false, singleMessageCredit, getDeliveryMethod, getRecordMethod);
}
queue.registerSubscription(sub,false);
@@ -180,27 +173,5 @@ public class BasicGetMethodHandler imple
}
- public static final class GetNoAckSubscription extends SubscriptionImpl.NoAckSubscription
- {
- public GetNoAckSubscription(AMQChannel channel, AMQProtocolSession protocolSession,
- AMQShortString consumerTag, FieldTable filters,
- boolean noLocal, FlowCreditManager creditManager,
- ClientDeliveryMethod deliveryMethod,
- RecordDeliveryMethod recordMethod)
- throws AMQException
- {
- super(channel, protocolSession, consumerTag, filters, noLocal, creditManager, deliveryMethod, recordMethod);
- }
-
- public boolean isTransient()
- {
- return true;
- }
- public boolean wouldSuspend(QueueEntry msg)
- {
- return !getCreditManager().useCreditForMessage(msg.getMessage());
- }
-
- }
}
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java Mon Sep 19 15:13:18 2011
@@ -68,5 +68,7 @@ public class ConnectionCloseMethodHandle
ConnectionCloseOkBody responseBody = methodRegistry.createConnectionCloseOkBody();
session.writeFrame(responseBody.generateFrame(channelId));
+ session.closeProtocolSession();
+
}
}
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java Mon Sep 19 15:13:18 2011
@@ -23,7 +23,6 @@ package org.apache.qpid.server.handler;
import javax.security.sasl.SaslException;
import javax.security.sasl.SaslServer;
-
import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.ConnectionCloseBody;
@@ -89,7 +88,10 @@ public class ConnectionSecureOkMethodHan
disposeSaslServer(session);
break;
case SUCCESS:
- _logger.info("Connected as: " + ss.getAuthorizationID());
+ if (_logger.isInfoEnabled())
+ {
+ _logger.info("Connected as: " + UsernamePrincipal.getUsernamePrincipalFromSubject(authResult.getSubject()));
+ }
stateManager.changeState(AMQState.CONNECTION_NOT_TUNED);
ConnectionTuneBody tuneBody =
@@ -97,8 +99,7 @@ public class ConnectionSecureOkMethodHan
ConnectionStartOkMethodHandler.getConfiguredFrameSize(),
ApplicationRegistry.getInstance().getConfiguration().getHeartBeatDelay());
session.writeFrame(tuneBody.generateFrame(0));
- final UsernamePrincipal principal = UsernamePrincipal.getUsernamePrincipalFromSubject(authResult.getSubject());
- session.setAuthorizedID(principal);
+ session.setAuthorizedSubject(authResult.getSubject());
disposeSaslServer(session);
break;
case CONTINUE:
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java Mon Sep 19 15:13:18 2011
@@ -65,7 +65,6 @@ public class ConnectionStartOkMethodHand
_logger.info("Locale selected: " + body.getLocale());
AuthenticationManager authMgr = ApplicationRegistry.getInstance().getAuthenticationManager();
-
SaslServer ss = null;
try
{
@@ -78,8 +77,7 @@ public class ConnectionStartOkMethodHand
session.setSaslServer(ss);
- AuthenticationResult authResult = authMgr.authenticate(ss, body.getResponse());
-
+ final AuthenticationResult authResult = authMgr.authenticate(ss, body.getResponse());
//save clientProperties
if (session.getClientProperties() == null)
{
@@ -108,8 +106,11 @@ public class ConnectionStartOkMethodHand
break;
case SUCCESS:
- _logger.info("Connected as: " + ss.getAuthorizationID());
- session.setAuthorizedID(new UsernamePrincipal(ss.getAuthorizationID()));
+ if (_logger.isInfoEnabled())
+ {
+ _logger.info("Connected as: " + UsernamePrincipal.getUsernamePrincipalFromSubject(authResult.getSubject()));
+ }
+ session.setAuthorizedSubject(authResult.getSubject());
stateManager.changeState(AMQState.CONNECTION_NOT_TUNED);
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java Mon Sep 19 15:13:18 2011
@@ -106,7 +106,7 @@ public class QueueDeclareHandler impleme
else
{
queue = createQueue(queueName, body, virtualHost, protocolConnection);
- queue.setPrincipalHolder(protocolConnection);
+ queue.setAuthorizationHolder(protocolConnection);
if (queue.isDurable() && !queue.isAutoDelete())
{
store.createQueue(queue, body.getArguments());
@@ -119,7 +119,7 @@ public class QueueDeclareHandler impleme
if (body.getExclusive())
{
queue.setExclusiveOwningSession(protocolConnection.getChannel(channelId));
- queue.setPrincipalHolder(protocolConnection);
+ queue.setAuthorizationHolder(protocolConnection);
if(!body.getDurable())
{
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/actors/ManagementActor.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/actors/ManagementActor.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/actors/ManagementActor.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/actors/ManagementActor.java Mon Sep 19 15:13:18 2011
@@ -20,11 +20,15 @@
*/
package org.apache.qpid.server.logging.actors;
-import org.apache.qpid.server.logging.LogMessage;
-import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.logging.RootMessageLogger;
+import java.security.AccessController;
+import java.security.Principal;
import java.text.MessageFormat;
+import java.util.Set;
+
+import javax.management.remote.JMXPrincipal;
+import javax.security.auth.Subject;
/**
* NOTE: This actor is not thread safe.
@@ -40,16 +44,23 @@ import java.text.MessageFormat;
*/
public class ManagementActor extends AbstractActor
{
+ /**
+ * Holds the principal name to display when principal subject is not available.
+ * <p>
+ * This is useful for cases when users invoke JMX operation over JConsole
+ * attached to the local JVM.
+ */
+ private static final String UNKNOWN_PRINCIPAL = "N/A";
+
String _lastThreadName = null;
/**
* LOG FORMAT for the ManagementActor,
- * Uses a MessageFormat call to insert the requried values according to
- * these indicies:
+ * Uses a MessageFormat call to insert the required values according to
+ * these indices:
*
- * 0 - Connection ID
- * 1 - User ID
- * 2 - IP
+ * 0 - User ID
+ * 1 - IP
*/
public static final String MANAGEMENT_FORMAT = "mng:{0}({1})";
@@ -75,19 +86,20 @@ public class ManagementActor extends Abs
_lastThreadName = currentName;
// Management Thread names have this format.
- //RMI TCP Connection(2)-169.24.29.116
+ // RMI TCP Connection(2)-169.24.29.116
// This is true for both LocalAPI and JMX Connections
// However to be defensive lets test.
String[] split = currentName.split("\\(");
if (split.length == 2)
{
- String connectionID = split[1].split("\\)")[0];
String ip = currentName.split("-")[1];
-
- actor = MessageFormat.format(MANAGEMENT_FORMAT,
- connectionID,
- ip);
+ String principalName = getPrincipalName();
+ if (principalName == null)
+ {
+ principalName = UNKNOWN_PRINCIPAL;
+ }
+ actor = MessageFormat.format(MANAGEMENT_FORMAT, principalName, ip);
}
else
{
@@ -105,6 +117,30 @@ public class ManagementActor extends Abs
}
}
+ /**
+ * Returns current JMX principal name.
+ *
+ * @return principal name or null if principal can not be found
+ */
+ protected String getPrincipalName()
+ {
+ String identity = null;
+
+ // retrieve Subject from current AccessControlContext
+ final Subject subject = Subject.getSubject(AccessController.getContext());
+ if (subject != null)
+ {
+ // retrieve JMXPrincipal from Subject
+ final Set<JMXPrincipal> principals = subject.getPrincipals(JMXPrincipal.class);
+ if (principals != null && !principals.isEmpty())
+ {
+ final Principal principal = principals.iterator().next();
+ identity = principal.getName();
+ }
+ }
+ return identity;
+ }
+
public String getLogMessage()
{
updateLogString();
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/Exchange_logmessages.properties
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/Exchange_logmessages.properties?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/Exchange_logmessages.properties (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/Exchange_logmessages.properties Mon Sep 19 15:13:18 2011
@@ -21,4 +21,5 @@
# 0 - type
# 1 - name
CREATED = EXH-1001 : Create :[ Durable] Type: {0} Name: {1}
-DELETED = EXH-1002 : Deleted
\ No newline at end of file
+DELETED = EXH-1002 : Deleted
+DISCARDMSG = EXH-1003 : Discarded Message : Name: {0} Routing Key: {1}
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/ManagementConsole_logmessages.properties
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/ManagementConsole_logmessages.properties?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/ManagementConsole_logmessages.properties (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/ManagementConsole_logmessages.properties Mon Sep 19 15:13:18 2011
@@ -30,4 +30,4 @@ STOPPED = MNG-1005 : Stopped
# 0 - Path
SSL_KEYSTORE = MNG-1006 : Using SSL Keystore : {0}
OPEN = MNG-1007 : Open : User {0}
-CLOSE = MNG-1008 : Close
\ No newline at end of file
+CLOSE = MNG-1008 : Close : User {0}
\ No newline at end of file
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java Mon Sep 19 15:13:18 2011
@@ -47,7 +47,7 @@ public class ChannelLogSubject extends A
*/
setLogStringWithFormat(CHANNEL_FORMAT,
session.getSessionID(),
- session.getPrincipal().getName(),
+ session.getAuthorizedPrincipal().getName(),
session.getRemoteAddress(),
session.getVirtualHost().getName(),
channel.getChannelId());
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java Mon Sep 19 15:13:18 2011
@@ -56,7 +56,7 @@ public class ConnectionLogSubject extend
{
if (!_upToDate)
{
- if (_session.getPrincipal() != null)
+ if (_session.getAuthorizedPrincipal() != null)
{
if (_session.getVirtualHost() != null)
{
@@ -72,7 +72,7 @@ public class ConnectionLogSubject extend
*/
_logString = "[" + MessageFormat.format(CONNECTION_FORMAT,
_session.getSessionID(),
- _session.getPrincipal().getName(),
+ _session.getAuthorizedPrincipal().getName(),
_session.getRemoteAddress(),
_session.getVirtualHost().getName())
+ "] ";
@@ -83,7 +83,7 @@ public class ConnectionLogSubject extend
{
_logString = "[" + MessageFormat.format(USER_FORMAT,
_session.getSessionID(),
- _session.getPrincipal().getName(),
+ _session.getAuthorizedPrincipal().getName(),
_session.getRemoteAddress())
+ "] ";
Propchange: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/management/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Mon Sep 19 15:13:18 2011
@@ -4,4 +4,4 @@
/qpid/branches/java-network-refactor/qpid/java/broker/src/main/java/org/apache/qpid/server/management:805429-821809
/qpid/branches/jmx_mc_gsoc09/qpid/java/broker/src/main/java/org/apache/qpid/server/management:787599
/qpid/branches/qpid-2935/qpid/java/broker/src/main/java/org/apache/qpid/server/management:1061302-1072333
-/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/management:753219-753220,753253,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790,1156188
+/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/management:753219-753220,753253,758730,759097,760919,761721,762365,762992,763959,764026,764109,764140,764790,1144319-1172654
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/management/JMXManagedObjectRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/management/JMXManagedObjectRegistry.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/management/JMXManagedObjectRegistry.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/management/JMXManagedObjectRegistry.java Mon Sep 19 15:13:18 2011
@@ -72,15 +72,14 @@ public class JMXManagedObjectRegistry im
{
private static final Logger _log = Logger.getLogger(JMXManagedObjectRegistry.class);
- public static final String MANAGEMENT_PORT_CONFIG_PATH = "management.jmxport";
- public static final int MANAGEMENT_PORT_DEFAULT = 8999;
- public static final int PORT_EXPORT_OFFSET = 100;
-
private final MBeanServer _mbeanServer;
private JMXConnectorServer _cs;
private Registry _rmiRegistry;
private boolean _useCustomSocketFactory;
+ private final int _jmxPortRegistryServer;
+ private final int _jmxPortConnectorServer;
+
public JMXManagedObjectRegistry() throws AMQException
{
_log.info("Initialising managed object registry using platform MBean server");
@@ -93,8 +92,11 @@ public class JMXManagedObjectRegistry im
_mbeanServer =
platformServer ? ManagementFactory.getPlatformMBeanServer()
: MBeanServerFactory.createMBeanServer(ManagedObject.DOMAIN);
- }
+ _jmxPortRegistryServer = appRegistry.getConfiguration().getJMXPortRegistryServer();
+ _jmxPortConnectorServer = appRegistry.getConfiguration().getJMXConnectorServerPort();
+
+ }
public void start() throws IOException, ConfigurationException
{
@@ -109,7 +111,6 @@ public class JMXManagedObjectRegistry im
}
IApplicationRegistry appRegistry = ApplicationRegistry.getInstance();
- int port = appRegistry.getConfiguration().getJMXManagementPort();
//Socket factories for the RMIConnectorServer, either default or SLL depending on configuration
@@ -204,14 +205,14 @@ public class JMXManagedObjectRegistry im
System.setProperty("java.rmi.server.randomIDs", "true");
if(_useCustomSocketFactory)
{
- _rmiRegistry = LocateRegistry.createRegistry(port, null, new CustomRMIServerSocketFactory());
+ _rmiRegistry = LocateRegistry.createRegistry(_jmxPortRegistryServer, null, new CustomRMIServerSocketFactory());
}
else
{
- _rmiRegistry = LocateRegistry.createRegistry(port, null, null);
+ _rmiRegistry = LocateRegistry.createRegistry(_jmxPortRegistryServer, null, null);
}
- CurrentActor.get().message(ManagementConsoleMessages.LISTENING("RMI Registry", port));
+ CurrentActor.get().message(ManagementConsoleMessages.LISTENING("RMI Registry", _jmxPortRegistryServer));
/*
* We must now create the RMI ConnectorServer manually, as the JMX Factory methods use RMI calls
@@ -222,7 +223,7 @@ public class JMXManagedObjectRegistry im
* The registry is exported on the defined management port 'port'. We will export the RMIConnectorServer
* on 'port +1'. Use of these two well-defined ports will ease any navigation through firewall's.
*/
- final RMIServerImpl rmiConnectorServerStub = new RMIJRMPServerImpl(port+PORT_EXPORT_OFFSET, csf, ssf, env);
+ final RMIServerImpl rmiConnectorServerStub = new RMIJRMPServerImpl(_jmxPortConnectorServer, csf, ssf, env);
String localHost;
try
{
@@ -234,9 +235,9 @@ public class JMXManagedObjectRegistry im
}
final String hostname = localHost;
final JMXServiceURL externalUrl = new JMXServiceURL(
- "service:jmx:rmi://"+hostname+":"+(port+PORT_EXPORT_OFFSET)+"/jndi/rmi://"+hostname+":"+port+"/jmxrmi");
+ "service:jmx:rmi://"+hostname+":"+(_jmxPortConnectorServer)+"/jndi/rmi://"+hostname+":"+_jmxPortRegistryServer+"/jmxrmi");
- final JMXServiceURL internalUrl = new JMXServiceURL("rmi", hostname, port+PORT_EXPORT_OFFSET);
+ final JMXServiceURL internalUrl = new JMXServiceURL("rmi", hostname, _jmxPortConnectorServer);
_cs = new RMIConnectorServer(internalUrl, env, rmiConnectorServerStub, _mbeanServer)
{
@Override
@@ -305,7 +306,7 @@ public class JMXManagedObjectRegistry im
_cs.start();
String connectorServer = (sslEnabled ? "SSL " : "") + "JMX RMIConnectorServer";
- CurrentActor.get().message(ManagementConsoleMessages.LISTENING(connectorServer, port + PORT_EXPORT_OFFSET));
+ CurrentActor.get().message(ManagementConsoleMessages.LISTENING(connectorServer, _jmxPortConnectorServer));
CurrentActor.get().message(ManagementConsoleMessages.READY(false));
}
@@ -400,7 +401,7 @@ public class JMXManagedObjectRegistry im
if (_rmiRegistry != null)
{
// Stopping the RMI registry
- CurrentActor.get().message(ManagementConsoleMessages.SHUTTING_DOWN("RMI Registry", _cs.getAddress().getPort() - PORT_EXPORT_OFFSET));
+ CurrentActor.get().message(ManagementConsoleMessages.SHUTTING_DOWN("RMI Registry", _jmxPortRegistryServer));
try
{
UnicastRemoteObject.unexportObject(_rmiRegistry, false);
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java Mon Sep 19 15:13:18 2011
@@ -26,7 +26,6 @@ import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.security.AccessControlContext;
import java.security.AccessController;
-import java.security.Principal;
import java.util.Set;
import javax.management.Attribute;
@@ -43,7 +42,6 @@ import javax.management.remote.MBeanServ
import javax.security.auth.Subject;
import org.apache.log4j.Logger;
-import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.actors.ManagementActor;
import org.apache.qpid.server.logging.messages.ManagementConsoleMessages;
import org.apache.qpid.server.registry.ApplicationRegistry;
@@ -51,17 +49,13 @@ import org.apache.qpid.server.security.S
import org.apache.qpid.server.security.access.Operation;
/**
- * This class can be used by the JMXConnectorServer as an InvocationHandler for the mbean operations. This implements
- * the logic for allowing the users to invoke MBean operations and implements the restrictions for readOnly, readWrite
- * and admin users.
+ * This class can be used by the JMXConnectorServer as an InvocationHandler for the mbean operations. It delegates
+ * JMX access decisions to the SecurityPlugin.
*/
public class MBeanInvocationHandlerImpl implements InvocationHandler, NotificationListener
{
private static final Logger _logger = Logger.getLogger(MBeanInvocationHandlerImpl.class);
- public final static String ADMIN = "admin";
- public final static String READWRITE = "readwrite";
- public final static String READONLY = "readonly";
private final static String DELEGATE = "JMImplementation:type=MBeanServerDelegate";
private MBeanServer _mbs;
private static ManagementActor _logActor;
@@ -135,14 +129,13 @@ public class MBeanInvocationHandlerImpl
Set<JMXPrincipal> principals = subject.getPrincipals(JMXPrincipal.class);
if (principals == null || principals.isEmpty())
{
- throw new SecurityException("Access denied: no principal");
+ throw new SecurityException("Access denied: no JMX principal");
}
-
- // Save the principal
- Principal principal = principals.iterator().next();
- SecurityManager.setThreadPrincipal(principal);
-
- // Get the component, type and impact, which may be null
+
+ // Save the subject
+ SecurityManager.setThreadSubject(subject);
+
+ // Get the component, type and impact, which may be null
String type = getType(method, args);
String vhost = getVirtualHost(method, args);
int impact = getImpact(method, args);
@@ -284,7 +277,7 @@ public class MBeanInvocationHandlerImpl
}
catch (JMException ex)
{
- ex.printStackTrace();
+ _logger.error("Unable to determine mbean impact for method : " + mbeanMethod, ex);
}
}
@@ -320,7 +313,7 @@ public class MBeanInvocationHandlerImpl
else if (notification.getType().equals(JMXConnectionNotification.CLOSED) ||
notification.getType().equals(JMXConnectionNotification.FAILED))
{
- _logActor.message(ManagementConsoleMessages.CLOSE());
+ _logActor.message(ManagementConsoleMessages.CLOSE(user));
}
}
}
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData.java Mon Sep 19 15:13:18 2011
@@ -29,7 +29,10 @@ import org.apache.qpid.framing.abstracti
import org.apache.qpid.server.store.StorableMessageMetaData;
import org.apache.qpid.server.store.MessageMetaDataType;
import org.apache.qpid.AMQException;
+import org.apache.qpid.server.util.ByteBufferInputStream;
+import org.apache.qpid.server.util.ByteBufferOutputStream;
+import java.io.*;
import java.nio.ByteBuffer;
import java.util.Set;
@@ -120,38 +123,38 @@ public class MessageMetaData implements
return size;
}
+
public int writeToBuffer(int offset, ByteBuffer dest)
{
- ByteBuffer src = ByteBuffer.allocate((int)getStorableSize());
-
- org.apache.mina.common.ByteBuffer minaSrc = org.apache.mina.common.ByteBuffer.wrap(src);
- EncodingUtils.writeInteger(minaSrc, _contentHeaderBody.getSize());
- _contentHeaderBody.writePayload(minaSrc);
- EncodingUtils.writeShortStringBytes(minaSrc, _messagePublishInfo.getExchange());
- EncodingUtils.writeShortStringBytes(minaSrc, _messagePublishInfo.getRoutingKey());
- byte flags = 0;
- if(_messagePublishInfo.isMandatory())
- {
- flags |= MANDATORY_FLAG;
- }
- if(_messagePublishInfo.isImmediate())
+ int oldPosition = dest.position();
+ try
{
- flags |= IMMEDIATE_FLAG;
+
+ DataOutputStream dataOutputStream = new DataOutputStream(new ByteBufferOutputStream(dest));
+ EncodingUtils.writeInteger(dataOutputStream, _contentHeaderBody.getSize());
+ _contentHeaderBody.writePayload(dataOutputStream);
+ EncodingUtils.writeShortStringBytes(dataOutputStream, _messagePublishInfo.getExchange());
+ EncodingUtils.writeShortStringBytes(dataOutputStream, _messagePublishInfo.getRoutingKey());
+ byte flags = 0;
+ if(_messagePublishInfo.isMandatory())
+ {
+ flags |= MANDATORY_FLAG;
+ }
+ if(_messagePublishInfo.isImmediate())
+ {
+ flags |= IMMEDIATE_FLAG;
+ }
+ dest.put(flags);
+ dest.putLong(_arrivalTime);
+
}
- EncodingUtils.writeByte(minaSrc, flags);
- EncodingUtils.writeLong(minaSrc,_arrivalTime);
- src.position(minaSrc.position());
- src.flip();
- src.position(offset);
- src = src.slice();
- if(dest.remaining() < src.limit())
+ catch (IOException e)
{
- src.limit(dest.remaining());
+ // This shouldn't happen as we are not actually using anything that can throw an IO Exception
+ throw new RuntimeException(e);
}
- dest.put(src);
-
- return src.limit();
+ return dest.position()-oldPosition;
}
public int getContentSize()
@@ -173,14 +176,15 @@ public class MessageMetaData implements
{
try
{
- org.apache.mina.common.ByteBuffer minaSrc = org.apache.mina.common.ByteBuffer.wrap(buf);
- int size = EncodingUtils.readInteger(minaSrc);
- ContentHeaderBody chb = ContentHeaderBody.createFromBuffer(minaSrc, size);
- final AMQShortString exchange = EncodingUtils.readAMQShortString(minaSrc);
- final AMQShortString routingKey = EncodingUtils.readAMQShortString(minaSrc);
+ ByteBufferInputStream bbis = new ByteBufferInputStream(buf);
+ DataInputStream dais = new DataInputStream(bbis);
+ int size = EncodingUtils.readInteger(dais);
+ ContentHeaderBody chb = ContentHeaderBody.createFromBuffer(dais, size);
+ final AMQShortString exchange = EncodingUtils.readAMQShortString(dais);
+ final AMQShortString routingKey = EncodingUtils.readAMQShortString(dais);
- final byte flags = EncodingUtils.readByte(minaSrc);
- long arrivalTime = EncodingUtils.readLong(minaSrc);
+ final byte flags = EncodingUtils.readByte(dais);
+ long arrivalTime = EncodingUtils.readLong(dais);
MessagePublishInfo publishBody =
new MessagePublishInfo()
@@ -216,6 +220,10 @@ public class MessageMetaData implements
{
throw new RuntimeException(e);
}
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
}
};
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData_0_10.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData_0_10.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData_0_10.java Mon Sep 19 15:13:18 2011
@@ -34,7 +34,7 @@ import org.apache.qpid.transport.codec.B
import java.nio.ByteBuffer;
import java.lang.ref.SoftReference;
-public class MessageMetaData_0_10 implements StorableMessageMetaData
+public class MessageMetaData_0_10 implements StorableMessageMetaData, InboundMessage
{
private Header _header;
private DeliveryProperties _deliveryProps;
@@ -194,6 +194,12 @@ public class MessageMetaData_0_10 implem
return _deliveryProps == null ? 0L : _deliveryProps.getExpiration();
}
+ public boolean isRedelivered()
+ {
+ // The *Message* is never redelivered, only queue entries are...
+ return false;
+ }
+
public long getArrivalTime()
{
return _arrivalTime;
@@ -239,4 +245,6 @@ public class MessageMetaData_0_10 implem
}
}
+
+
}
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java Mon Sep 19 15:13:18 2011
@@ -26,6 +26,7 @@
*/
package org.apache.qpid.server.output.amqp0_8;
+import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.message.AMQMessage;
import org.apache.qpid.server.queue.QueueEntry;
@@ -34,22 +35,18 @@ import org.apache.qpid.server.output.Hea
import org.apache.qpid.server.message.MessageContentSource;
import org.apache.qpid.server.message.MessageTransferMessage;
import org.apache.qpid.framing.*;
-import org.apache.qpid.framing.amqp_8_0.BasicGetBodyImpl;
import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.framing.abstraction.ProtocolVersionMethodConverter;
import org.apache.qpid.AMQException;
import org.apache.qpid.transport.DeliveryProperties;
-import java.nio.ByteBuffer;
+import java.io.DataOutputStream;
+import java.io.IOException;
public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
{
private static final MethodRegistry METHOD_REGISTRY = MethodRegistry.getMethodRegistry(ProtocolVersion.v8_0);
- private static final ProtocolVersionMethodConverter PROTOCOL_CONVERTER =
- METHOD_REGISTRY.getProtocolVersionMethodConverter();
-
public static Factory getInstanceFactory()
{
return new Factory()
@@ -62,6 +59,7 @@ public class ProtocolOutputConverterImpl
};
}
+
private final AMQProtocolSession _protocolSession;
private ProtocolOutputConverterImpl(AMQProtocolSession session)
@@ -78,10 +76,11 @@ public class ProtocolOutputConverterImpl
public void writeDeliver(QueueEntry entry, int channelId, long deliveryTag, AMQShortString consumerTag)
throws AMQException
{
- AMQDataBlock deliver = createEncodedDeliverFrame(entry, channelId, deliveryTag, consumerTag);
- writeMessageDelivery(entry.getMessage(), getContentHeaderBody(entry), channelId, deliver);
+ AMQBody deliverBody = createEncodedDeliverBody(entry, deliveryTag, consumerTag);
+ writeMessageDelivery(entry, channelId, deliverBody);
}
+
private ContentHeaderBody getContentHeaderBody(QueueEntry entry)
throws AMQException
{
@@ -93,65 +92,120 @@ public class ProtocolOutputConverterImpl
{
final MessageTransferMessage message = (MessageTransferMessage) entry.getMessage();
BasicContentHeaderProperties props = HeaderPropertiesConverter.convert(message);
- ContentHeaderBody chb = new ContentHeaderBody(props, BasicGetBodyImpl.CLASS_ID);
- chb.bodySize = message.getSize();
+ ContentHeaderBody chb = new ContentHeaderBody(props, org.apache.qpid.framing.amqp_8_0.BasicGetBodyImpl.CLASS_ID);
+ chb.bodySize = message.getSize();
return chb;
}
}
- public void writeGetOk(QueueEntry entry, int channelId, long deliveryTag, int queueSize) throws AMQException
+ private void writeMessageDelivery(QueueEntry entry, int channelId, AMQBody deliverBody)
+ throws AMQException
{
- AMQDataBlock deliver = createEncodedGetOkFrame(entry, channelId, deliveryTag, queueSize);
- writeMessageDelivery(entry.getMessage(), getContentHeaderBody(entry), channelId, deliver);
+ writeMessageDelivery(entry.getMessage(), getContentHeaderBody(entry), channelId, deliverBody);
}
- private void writeMessageDelivery(MessageContentSource message, ContentHeaderBody chb, int channelId, AMQDataBlock deliver)
+ private void writeMessageDelivery(MessageContentSource message, ContentHeaderBody contentHeaderBody, int channelId, AMQBody deliverBody)
throws AMQException
{
- AMQDataBlock contentHeader = ContentHeaderBody.createAMQFrame(channelId, chb);
-
+ int bodySize = (int) message.getSize();
- final int bodySize = (int) message.getSize();
if(bodySize == 0)
{
- SmallCompositeAMQDataBlock compositeBlock = new SmallCompositeAMQDataBlock(deliver,
- contentHeader);
+ SmallCompositeAMQBodyBlock compositeBlock = new SmallCompositeAMQBodyBlock(channelId, deliverBody,
+ contentHeaderBody);
+
writeFrame(compositeBlock);
}
else
{
int maxBodySize = (int) getProtocolSession().getMaxFrameSize() - AMQFrame.getFrameOverhead();
- final int capacity = bodySize > maxBodySize ? maxBodySize : bodySize;
- ByteBuffer buf = ByteBuffer.allocate(capacity);
- int writtenSize = 0;
+ int capacity = bodySize > maxBodySize ? maxBodySize : bodySize;
+
+ int writtenSize = capacity;
+
+ AMQBody firstContentBody = new MessageContentSourceBody(message,0,capacity);
- writtenSize += message.getContent(buf, writtenSize);
- buf.flip();
- AMQDataBlock firstContentBody = new AMQFrame(channelId, PROTOCOL_CONVERTER.convertToBody(buf));
- AMQDataBlock[] blocks = new AMQDataBlock[]{deliver, contentHeader, firstContentBody};
- CompositeAMQDataBlock compositeBlock = new CompositeAMQDataBlock(blocks);
+ CompositeAMQBodyBlock
+ compositeBlock = new CompositeAMQBodyBlock(channelId, deliverBody, contentHeaderBody, firstContentBody);
writeFrame(compositeBlock);
while(writtenSize < bodySize)
{
- buf = java.nio.ByteBuffer.allocate(capacity);
- writtenSize += message.getContent(buf, writtenSize);
- buf.flip();
- writeFrame(new AMQFrame(channelId, PROTOCOL_CONVERTER.convertToBody(buf)));
+ capacity = bodySize - writtenSize > maxBodySize ? maxBodySize : bodySize - writtenSize;
+ MessageContentSourceBody body = new MessageContentSourceBody(message, writtenSize, capacity);
+ writtenSize += capacity;
+
+ writeFrame(new AMQFrame(channelId, body));
}
+ }
+ }
+ private class MessageContentSourceBody implements AMQBody
+ {
+ public static final byte TYPE = 3;
+ private int _length;
+ private MessageContentSource _message;
+ private int _offset;
+
+ public MessageContentSourceBody(MessageContentSource message, int offset, int length)
+ {
+ _message = message;
+ _offset = offset;
+ _length = length;
+ }
+
+ public byte getFrameType()
+ {
+ return TYPE;
}
+
+ public int getSize()
+ {
+ return _length;
+ }
+
+ public void writePayload(DataOutputStream buffer) throws IOException
+ {
+ byte[] data = new byte[_length];
+
+ _message.getContent(java.nio.ByteBuffer.wrap(data), _offset);
+
+ buffer.write(data);
+ }
+
+ public void handle(int channelId, AMQVersionAwareProtocolSession amqProtocolSession) throws AMQException
+ {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ private AMQDataBlock createContentHeaderBlock(final int channelId, final ContentHeaderBody contentHeaderBody)
+ {
+
+ AMQDataBlock contentHeader = ContentHeaderBody.createAMQFrame(channelId,
+ contentHeaderBody);
+ return contentHeader;
}
- private AMQDataBlock createEncodedDeliverFrame(QueueEntry entry, int channelId, long deliveryTag, AMQShortString consumerTag)
+ public void writeGetOk(QueueEntry entry, int channelId, long deliveryTag, int queueSize) throws AMQException
+ {
+ AMQBody deliver = createEncodedGetOkBody(entry, deliveryTag, queueSize);
+ writeMessageDelivery(entry, channelId, deliver);
+ }
+
+
+ private AMQBody createEncodedDeliverBody(QueueEntry entry,
+ final long deliveryTag,
+ final AMQShortString consumerTag)
throws AMQException
{
+
final AMQShortString exchangeName;
final AMQShortString routingKey;
@@ -172,21 +226,58 @@ public class ProtocolOutputConverterImpl
final boolean isRedelivered = entry.isRedelivered();
+ final AMQBody returnBlock = new AMQBody()
+ {
+
+ public AMQBody _underlyingBody;
- BasicDeliverBody deliverBody =
- METHOD_REGISTRY.createBasicDeliverBody(consumerTag,
- deliveryTag,
- isRedelivered,
- exchangeName,
- routingKey);
+ public AMQBody createAMQBody()
+ {
+ return METHOD_REGISTRY.createBasicDeliverBody(consumerTag,
+ deliveryTag,
+ isRedelivered,
+ exchangeName,
+ routingKey);
- AMQFrame deliverFrame = deliverBody.generateFrame(channelId);
- return deliverFrame;
+
+
+ }
+
+ public byte getFrameType()
+ {
+ return AMQMethodBody.TYPE;
+ }
+
+ public int getSize()
+ {
+ if(_underlyingBody == null)
+ {
+ _underlyingBody = createAMQBody();
+ }
+ return _underlyingBody.getSize();
+ }
+
+ public void writePayload(DataOutputStream buffer) throws IOException
+ {
+ if(_underlyingBody == null)
+ {
+ _underlyingBody = createAMQBody();
+ }
+ _underlyingBody.writePayload(buffer);
+ }
+
+ public void handle(final int channelId, final AMQVersionAwareProtocolSession amqMinaProtocolSession)
+ throws AMQException
+ {
+ throw new AMQException("This block should never be dispatched!");
+ }
+ };
+ return returnBlock;
}
- private AMQDataBlock createEncodedGetOkFrame(QueueEntry entry, int channelId, long deliveryTag, int queueSize)
+ private AMQBody createEncodedGetOkBody(QueueEntry entry, long deliveryTag, int queueSize)
throws AMQException
{
final AMQShortString exchangeName;
@@ -215,9 +306,8 @@ public class ProtocolOutputConverterImpl
exchangeName,
routingKey,
queueSize);
- AMQFrame getOkFrame = getOkBody.generateFrame(channelId);
- return getOkFrame;
+ return getOkBody;
}
public byte getProtocolMinorVersion()
@@ -230,31 +320,28 @@ public class ProtocolOutputConverterImpl
return getProtocolSession().getProtocolMajorVersion();
}
- private AMQDataBlock createEncodedReturnFrame(MessagePublishInfo messagePublishInfo, int channelId, int replyCode, AMQShortString replyText) throws AMQException
+ private AMQBody createEncodedReturnFrame(MessagePublishInfo messagePublishInfo,
+ int replyCode,
+ AMQShortString replyText) throws AMQException
{
+
BasicReturnBody basicReturnBody =
METHOD_REGISTRY.createBasicReturnBody(replyCode,
- replyText,
- messagePublishInfo.getExchange(),
- messagePublishInfo.getRoutingKey());
- AMQFrame returnFrame = basicReturnBody.generateFrame(channelId);
+ replyText,
+ messagePublishInfo.getExchange(),
+ messagePublishInfo.getRoutingKey());
- return returnFrame;
+
+ return basicReturnBody;
}
- public void writeReturn(MessagePublishInfo messagePublishInfo,
- ContentHeaderBody header,
- MessageContentSource content,
- int channelId,
- int replyCode,
- AMQShortString replyText)
+ public void writeReturn(MessagePublishInfo messagePublishInfo, ContentHeaderBody header, MessageContentSource message, int channelId, int replyCode, AMQShortString replyText)
throws AMQException
{
- AMQDataBlock returnFrame = createEncodedReturnFrame(messagePublishInfo, channelId, replyCode, replyText);
-
- writeMessageDelivery(content, header, channelId, returnFrame);
+ AMQBody returnFrame = createEncodedReturnFrame(messagePublishInfo, replyCode, replyText);
+ writeMessageDelivery(message, header, channelId, returnFrame);
}
@@ -266,8 +353,68 @@ public class ProtocolOutputConverterImpl
public void confirmConsumerAutoClose(int channelId, AMQShortString consumerTag)
{
+
BasicCancelOkBody basicCancelOkBody = METHOD_REGISTRY.createBasicCancelOkBody(consumerTag);
writeFrame(basicCancelOkBody.generateFrame(channelId));
}
+
+
+ public static final class CompositeAMQBodyBlock extends AMQDataBlock
+ {
+ public static final int OVERHEAD = 3 * AMQFrame.getFrameOverhead();
+
+ private final AMQBody _methodBody;
+ private final AMQBody _headerBody;
+ private final AMQBody _contentBody;
+ private final int _channel;
+
+
+ public CompositeAMQBodyBlock(int channel, AMQBody methodBody, AMQBody headerBody, AMQBody contentBody)
+ {
+ _channel = channel;
+ _methodBody = methodBody;
+ _headerBody = headerBody;
+ _contentBody = contentBody;
+
+ }
+
+ public long getSize()
+ {
+ return OVERHEAD + _methodBody.getSize() + _headerBody.getSize() + _contentBody.getSize();
+ }
+
+ public void writePayload(DataOutputStream buffer) throws IOException
+ {
+ AMQFrame.writeFrames(buffer, _channel, _methodBody, _headerBody, _contentBody);
+ }
+ }
+
+ public static final class SmallCompositeAMQBodyBlock extends AMQDataBlock
+ {
+ public static final int OVERHEAD = 2 * AMQFrame.getFrameOverhead();
+
+ private final AMQBody _methodBody;
+ private final AMQBody _headerBody;
+ private final int _channel;
+
+
+ public SmallCompositeAMQBodyBlock(int channel, AMQBody methodBody, AMQBody headerBody)
+ {
+ _channel = channel;
+ _methodBody = methodBody;
+ _headerBody = headerBody;
+
+ }
+
+ public long getSize()
+ {
+ return OVERHEAD + _methodBody.getSize() + _headerBody.getSize() ;
+ }
+
+ public void writePayload(DataOutputStream buffer) throws IOException
+ {
+ AMQFrame.writeFrames(buffer, _channel, _methodBody, _headerBody);
+ }
+ }
}
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java Mon Sep 19 15:13:18 2011
@@ -20,9 +20,6 @@ package org.apache.qpid.server.output.am
*
*/
-
-import org.apache.mina.common.ByteBuffer;
-
import org.apache.qpid.server.output.ProtocolOutputConverter;
import org.apache.qpid.server.output.HeaderPropertiesConverter;
import org.apache.qpid.server.protocol.AMQProtocolSession;
@@ -38,11 +35,13 @@ import org.apache.qpid.AMQException;
import org.apache.qpid.transport.DeliveryProperties;
import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
{
private static final MethodRegistry METHOD_REGISTRY = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9);
- private static final ProtocolVersionMethodConverter
- PROTOCOL_CONVERTER = METHOD_REGISTRY.getProtocolVersionMethodConverter();
public static Factory getInstanceFactory()
@@ -121,15 +120,12 @@ public class ProtocolOutputConverterImpl
int maxBodySize = (int) getProtocolSession().getMaxFrameSize() - AMQFrame.getFrameOverhead();
- final int capacity = bodySize > maxBodySize ? maxBodySize : bodySize;
- java.nio.ByteBuffer buf = java.nio.ByteBuffer.allocate(capacity);
+ int capacity = bodySize > maxBodySize ? maxBodySize : bodySize;
- int writtenSize = 0;
+ int writtenSize = capacity;
+ AMQBody firstContentBody = new MessageContentSourceBody(message,0,capacity);
- writtenSize += message.getContent(buf, writtenSize);
- buf.flip();
- AMQBody firstContentBody = PROTOCOL_CONVERTER.convertToBody(buf);
CompositeAMQBodyBlock
compositeBlock = new CompositeAMQBodyBlock(channelId, deliverBody, contentHeaderBody, firstContentBody);
@@ -137,15 +133,55 @@ public class ProtocolOutputConverterImpl
while(writtenSize < bodySize)
{
- buf = java.nio.ByteBuffer.allocate(capacity);
+ capacity = bodySize - writtenSize > maxBodySize ? maxBodySize : bodySize - writtenSize;
+ MessageContentSourceBody body = new MessageContentSourceBody(message, writtenSize, capacity);
+ writtenSize += capacity;
- writtenSize += message.getContent(buf, writtenSize);
- buf.flip();
- writeFrame(new AMQFrame(channelId, PROTOCOL_CONVERTER.convertToBody(buf)));
+ writeFrame(new AMQFrame(channelId, body));
}
}
}
+ private class MessageContentSourceBody implements AMQBody
+ {
+ public static final byte TYPE = 3;
+ private int _length;
+ private MessageContentSource _message;
+ private int _offset;
+
+ public MessageContentSourceBody(MessageContentSource message, int offset, int length)
+ {
+ _message = message;
+ _offset = offset;
+ _length = length;
+ }
+
+ public byte getFrameType()
+ {
+ return TYPE;
+ }
+
+ public int getSize()
+ {
+ return _length;
+ }
+
+ public void writePayload(DataOutputStream buffer) throws IOException
+ {
+ byte[] data = new byte[_length];
+
+ _message.getContent(ByteBuffer.wrap(data), _offset);
+
+ buffer.write(data);
+ }
+
+ public void handle(int channelId, AMQVersionAwareProtocolSession amqProtocolSession) throws AMQException
+ {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+
private AMQDataBlock createContentHeaderBlock(final int channelId, final ContentHeaderBody contentHeaderBody)
{
@@ -221,7 +257,7 @@ public class ProtocolOutputConverterImpl
return _underlyingBody.getSize();
}
- public void writePayload(ByteBuffer buffer)
+ public void writePayload(DataOutputStream buffer) throws IOException
{
if(_underlyingBody == null)
{
@@ -346,7 +382,7 @@ public class ProtocolOutputConverterImpl
return OVERHEAD + _methodBody.getSize() + _headerBody.getSize() + _contentBody.getSize();
}
- public void writePayload(ByteBuffer buffer)
+ public void writePayload(DataOutputStream buffer) throws IOException
{
AMQFrame.writeFrames(buffer, _channel, _methodBody, _headerBody, _contentBody);
}
@@ -374,7 +410,7 @@ public class ProtocolOutputConverterImpl
return OVERHEAD + _methodBody.getSize() + _headerBody.getSize() ;
}
- public void writePayload(ByteBuffer buffer)
+ public void writePayload(DataOutputStream buffer) throws IOException
{
AMQFrame.writeFrames(buffer, _channel, _methodBody, _headerBody);
}
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9_1/ProtocolOutputConverterImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9_1/ProtocolOutputConverterImpl.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9_1/ProtocolOutputConverterImpl.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9_1/ProtocolOutputConverterImpl.java Mon Sep 19 15:13:18 2011
@@ -20,9 +20,6 @@ package org.apache.qpid.server.output.am
*
*/
-
-import org.apache.mina.common.ByteBuffer;
-
import org.apache.qpid.server.output.ProtocolOutputConverter;
import org.apache.qpid.server.output.HeaderPropertiesConverter;
import org.apache.qpid.server.protocol.AMQProtocolSession;
@@ -33,17 +30,16 @@ import org.apache.qpid.server.message.Me
import org.apache.qpid.framing.*;
import org.apache.qpid.framing.amqp_0_91.BasicGetBodyImpl;
import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.framing.abstraction.ProtocolVersionMethodConverter;
import org.apache.qpid.AMQException;
import org.apache.qpid.transport.DeliveryProperties;
import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
{
private static final MethodRegistry METHOD_REGISTRY = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_91);
- private static final ProtocolVersionMethodConverter
- PROTOCOL_CONVERTER = METHOD_REGISTRY.getProtocolVersionMethodConverter();
-
public static Factory getInstanceFactory()
{
@@ -121,15 +117,11 @@ public class ProtocolOutputConverterImpl
int maxBodySize = (int) getProtocolSession().getMaxFrameSize() - AMQFrame.getFrameOverhead();
- final int capacity = bodySize > maxBodySize ? maxBodySize : bodySize;
- java.nio.ByteBuffer buf = java.nio.ByteBuffer.allocate(capacity);
+ int capacity = bodySize > maxBodySize ? maxBodySize : bodySize;
- int writtenSize = 0;
+ int writtenSize = capacity;
-
- writtenSize += message.getContent(buf, writtenSize);
- buf.flip();
- AMQBody firstContentBody = PROTOCOL_CONVERTER.convertToBody(buf);
+ AMQBody firstContentBody = new MessageContentSourceBody(message,0,capacity);
CompositeAMQBodyBlock
compositeBlock = new CompositeAMQBodyBlock(channelId, deliverBody, contentHeaderBody, firstContentBody);
@@ -137,15 +129,54 @@ public class ProtocolOutputConverterImpl
while(writtenSize < bodySize)
{
- buf = java.nio.ByteBuffer.allocate(capacity);
+ capacity = bodySize - writtenSize > maxBodySize ? maxBodySize : bodySize - writtenSize;
+ MessageContentSourceBody body = new MessageContentSourceBody(message, writtenSize, capacity);
+ writtenSize += capacity;
- writtenSize += message.getContent(buf, writtenSize);
- buf.flip();
- writeFrame(new AMQFrame(channelId, PROTOCOL_CONVERTER.convertToBody(buf)));
+ writeFrame(new AMQFrame(channelId, body));
}
}
}
+ private class MessageContentSourceBody implements AMQBody
+ {
+ public static final byte TYPE = 3;
+ private int _length;
+ private MessageContentSource _message;
+ private int _offset;
+
+ public MessageContentSourceBody(MessageContentSource message, int offset, int length)
+ {
+ _message = message;
+ _offset = offset;
+ _length = length;
+ }
+
+ public byte getFrameType()
+ {
+ return TYPE;
+ }
+
+ public int getSize()
+ {
+ return _length;
+ }
+
+ public void writePayload(DataOutputStream buffer) throws IOException
+ {
+ byte[] data = new byte[_length];
+
+ _message.getContent(java.nio.ByteBuffer.wrap(data), _offset);
+
+ buffer.write(data);
+ }
+
+ public void handle(int channelId, AMQVersionAwareProtocolSession amqProtocolSession) throws AMQException
+ {
+ throw new UnsupportedOperationException();
+ }
+ }
+
private AMQDataBlock createContentHeaderBlock(final int channelId, final ContentHeaderBody contentHeaderBody)
{
@@ -221,7 +252,7 @@ public class ProtocolOutputConverterImpl
return _underlyingBody.getSize();
}
- public void writePayload(ByteBuffer buffer)
+ public void writePayload(DataOutputStream buffer) throws IOException
{
if(_underlyingBody == null)
{
@@ -346,7 +377,7 @@ public class ProtocolOutputConverterImpl
return OVERHEAD + _methodBody.getSize() + _headerBody.getSize() + _contentBody.getSize();
}
- public void writePayload(ByteBuffer buffer)
+ public void writePayload(DataOutputStream buffer) throws IOException
{
AMQFrame.writeFrames(buffer, _channel, _methodBody, _headerBody, _contentBody);
}
@@ -374,7 +405,7 @@ public class ProtocolOutputConverterImpl
return OVERHEAD + _methodBody.getSize() + _headerBody.getSize() ;
}
- public void writePayload(ByteBuffer buffer)
+ public void writePayload(DataOutputStream buffer) throws IOException
{
AMQFrame.writeFrames(buffer, _channel, _methodBody, _headerBody);
}
Modified: qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/Plugin.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/Plugin.java?rev=1172657&r1=1172656&r2=1172657&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/Plugin.java (original)
+++ qpid/branches/qpid-3346/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/Plugin.java Mon Sep 19 15:13:18 2011
@@ -27,5 +27,5 @@ public interface Plugin
/**
* Provide Configuration to this plugin
*/
- public void configure(ConfigurationPlugin config);
+ public void configure(ConfigurationPlugin config) throws ConfigurationException;
}
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org