You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rh...@apache.org on 2006/09/20 00:07:25 UTC
svn commit: r447994 [15/46] - in /incubator/qpid/trunk/qpid: ./ cpp/
cpp/bin/ cpp/broker/ cpp/broker/inc/ cpp/broker/src/ cpp/broker/test/
cpp/client/ cpp/client/inc/ cpp/client/src/ cpp/client/test/ cpp/common/
cpp/common/concurrent/ cpp/common/concur...
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/UsernamePasswordInitialiser.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/UsernamePasswordInitialiser.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/UsernamePasswordInitialiser.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/UsernamePasswordInitialiser.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,99 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.security.auth;
+
+import org.apache.commons.configuration.Configuration;
+
+import javax.security.auth.callback.*;
+import javax.security.auth.login.AccountNotFoundException;
+import javax.security.sasl.AuthorizeCallback;
+import java.util.Map;
+import java.io.IOException;
+import java.security.Principal;
+
+public abstract class UsernamePasswordInitialiser implements AuthenticationProviderInitialiser
+{
+ private ServerCallbackHandler _callbackHandler;
+
+ private class ServerCallbackHandler implements CallbackHandler
+ {
+ private final PrincipalDatabase _principalDatabase;
+
+ protected ServerCallbackHandler(PrincipalDatabase database)
+ {
+ _principalDatabase = database;
+ }
+
+ public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException
+ {
+ Principal username = null;
+ for (Callback callback : callbacks)
+ {
+ if (callback instanceof NameCallback)
+ {
+ username = new UsernamePrincipal(((NameCallback)callback).getDefaultName());
+ }
+ else if (callback instanceof PasswordCallback)
+ {
+ try
+ {
+ _principalDatabase.setPassword(username, (PasswordCallback) callback);
+ }
+ catch (AccountNotFoundException e)
+ {
+ // very annoyingly the callback handler does not throw anything more appropriate than
+ // IOException
+ throw new IOException("Error looking up user " + e);
+ }
+ }
+ else if (callback instanceof AuthorizeCallback)
+ {
+ ((AuthorizeCallback)callback).setAuthorized(true);
+ }
+ else
+ {
+ throw new UnsupportedCallbackException(callback);
+ }
+ }
+ }
+ }
+
+ public void initialise(String baseConfigPath, Configuration configuration,
+ Map<String, PrincipalDatabase> principalDatabases) throws Exception
+ {
+ String principalDatabaseName = configuration.getString(baseConfigPath + ".principal-database");
+ PrincipalDatabase db = principalDatabases.get(principalDatabaseName);
+ if (db == null)
+ {
+ throw new Exception("Principal database " + principalDatabaseName + " not found. Ensure the name matches " +
+ "an entry in the configuration file");
+ }
+ _callbackHandler = new ServerCallbackHandler(db);
+ }
+
+ public CallbackHandler getCallbackHandler()
+ {
+ return _callbackHandler;
+ }
+
+ public Map<String, ?> getProperties()
+ {
+ // there are no properties required for the CRAM-MD5 implementation
+ return null;
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/UsernamePasswordInitialiser.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/UsernamePrincipal.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/UsernamePrincipal.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/UsernamePrincipal.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/UsernamePrincipal.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,39 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.security.auth;
+
+import java.security.Principal;
+
+/**
+ * A principal that is just a wrapper for a simple username.
+ *
+ */
+public class UsernamePrincipal implements Principal
+{
+ private String _name;
+
+ public UsernamePrincipal(String name)
+ {
+ _name = name;
+ }
+
+ public String getName()
+ {
+ return _name;
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/UsernamePrincipal.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainInitialiser.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainInitialiser.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainInitialiser.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainInitialiser.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,35 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.security.auth.amqplain;
+
+import org.apache.qpid.server.security.auth.UsernamePasswordInitialiser;
+
+import javax.security.sasl.SaslServerFactory;
+
+public class AmqPlainInitialiser extends UsernamePasswordInitialiser
+{
+ public String getMechanismName()
+ {
+ return "AMQPLAIN";
+ }
+
+ public Class<? extends SaslServerFactory> getServerFactoryClassForJCARegistration()
+ {
+ return AmqPlainSaslServerFactory.class;
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainInitialiser.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainSaslServer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainSaslServer.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainSaslServer.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainSaslServer.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,120 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.security.auth.amqplain;
+
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.framing.AMQFrameDecodingException;
+import org.apache.mina.common.ByteBuffer;
+
+import javax.security.sasl.SaslServer;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.AuthorizeCallback;
+import javax.security.auth.callback.*;
+import java.io.IOException;
+
+public class AmqPlainSaslServer implements SaslServer
+{
+ public static final String MECHANISM = "AMQPLAIN";
+
+ private CallbackHandler _cbh;
+
+ private String _authorizationId;
+
+ private boolean _complete = false;
+
+ public AmqPlainSaslServer(CallbackHandler cbh)
+ {
+ _cbh = cbh;
+ }
+
+ public String getMechanismName()
+ {
+ return MECHANISM;
+ }
+
+ public byte[] evaluateResponse(byte[] response) throws SaslException
+ {
+ try
+ {
+ final FieldTable ft = new FieldTable(ByteBuffer.wrap(response), response.length);
+ String username = (String) ft.get("LOGIN");
+ // we do not care about the prompt but it throws if null
+ NameCallback nameCb = new NameCallback("prompt", username);
+ // we do not care about the prompt but it throws if null
+ PasswordCallback passwordCb = new PasswordCallback("prompt", false);
+ // TODO: should not get pwd as a String but as a char array...
+ String pwd = (String) ft.get("PASSWORD");
+ passwordCb.setPassword(pwd.toCharArray());
+ AuthorizeCallback authzCb = new AuthorizeCallback(username, username);
+ Callback[] callbacks = new Callback[]{nameCb, passwordCb, authzCb};
+ _cbh.handle(callbacks);
+ _complete = true;
+ if (authzCb.isAuthorized())
+ {
+ _authorizationId = authzCb.getAuthenticationID();
+ return null;
+ }
+ else
+ {
+ throw new SaslException("Authentication failed");
+ }
+ }
+ catch (AMQFrameDecodingException e)
+ {
+ throw new SaslException("Unable to decode response: " + e, e);
+ }
+ catch (IOException e)
+ {
+ throw new SaslException("Error processing data: " + e, e);
+ }
+ catch (UnsupportedCallbackException e)
+ {
+ throw new SaslException("Unable to obtain data from callback handler: " + e, e);
+ }
+ }
+
+ public boolean isComplete()
+ {
+ return _complete;
+ }
+
+ public String getAuthorizationID()
+ {
+ return _authorizationId;
+ }
+
+ public byte[] unwrap(byte[] incoming, int offset, int len) throws SaslException
+ {
+ throw new SaslException("Unsupported operation");
+ }
+
+ public byte[] wrap(byte[] outgoing, int offset, int len) throws SaslException
+ {
+ throw new SaslException("Unsupported operation");
+ }
+
+ public Object getNegotiatedProperty(String propName)
+ {
+ return null;
+ }
+
+ public void dispose() throws SaslException
+ {
+ _cbh = null;
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainSaslServer.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainSaslServerFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainSaslServerFactory.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainSaslServerFactory.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainSaslServerFactory.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,56 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.security.auth.amqplain;
+
+import javax.security.sasl.SaslServerFactory;
+import javax.security.sasl.SaslServer;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.Sasl;
+import javax.security.auth.callback.CallbackHandler;
+import java.util.Map;
+
+public class AmqPlainSaslServerFactory implements SaslServerFactory
+{
+ public SaslServer createSaslServer(String mechanism, String protocol, String serverName, Map props,
+ CallbackHandler cbh) throws SaslException
+ {
+ if (AmqPlainSaslServer.MECHANISM.equals(mechanism))
+ {
+ return new AmqPlainSaslServer(cbh);
+ }
+ else
+ {
+ return null;
+ }
+ }
+
+ public String[] getMechanismNames(Map props)
+ {
+ if (props.containsKey(Sasl.POLICY_NOPLAINTEXT) ||
+ props.containsKey(Sasl.POLICY_NODICTIONARY) ||
+ props.containsKey(Sasl.POLICY_NOACTIVE))
+ {
+ // returned array must be non null according to interface documentation
+ return new String[0];
+ }
+ else
+ {
+ return new String[]{AmqPlainSaslServer.MECHANISM};
+ }
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/amqplain/AmqPlainSaslServerFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainInitialiser.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainInitialiser.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainInitialiser.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainInitialiser.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,35 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.security.auth.plain;
+
+import org.apache.qpid.server.security.auth.UsernamePasswordInitialiser;
+
+import javax.security.sasl.SaslServerFactory;
+
+public class PlainInitialiser extends UsernamePasswordInitialiser
+{
+ public String getMechanismName()
+ {
+ return "PLAIN";
+ }
+
+ public Class<? extends SaslServerFactory> getServerFactoryClassForJCARegistration()
+ {
+ return PlainSaslServerFactory.class;
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainInitialiser.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainSaslServer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainSaslServer.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainSaslServer.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainSaslServer.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,141 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.security.auth.plain;
+
+import javax.security.auth.callback.*;
+import javax.security.sasl.AuthorizeCallback;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import java.io.IOException;
+
+public class PlainSaslServer implements SaslServer
+{
+ public static final String MECHANISM = "PLAIN";
+
+ private CallbackHandler _cbh;
+
+ private String _authorizationId;
+
+ private boolean _complete = false;
+
+ public PlainSaslServer(CallbackHandler cbh)
+ {
+ _cbh = cbh;
+ }
+
+ public String getMechanismName()
+ {
+ return MECHANISM;
+ }
+
+ public byte[] evaluateResponse(byte[] response) throws SaslException
+ {
+ try
+ {
+ int authzidNullPosition = findNullPosition(response, 0);
+ if (authzidNullPosition < 0)
+ {
+ throw new SaslException("Invalid PLAIN encoding, authzid null terminator not found");
+ }
+ int authcidNullPosition = findNullPosition(response, authzidNullPosition + 1);
+ if (authcidNullPosition < 0)
+ {
+ throw new SaslException("Invalid PLAIN encoding, authcid null terminator not found");
+ }
+
+ // we do not currently support authcid in any meaningful way
+ String authcid = new String(response, 0, authzidNullPosition, "utf8");
+ String authzid = new String(response, authzidNullPosition + 1, authcidNullPosition - 1, "utf8");
+
+ // we do not care about the prompt but it throws if null
+ NameCallback nameCb = new NameCallback("prompt", authzid);
+ // we do not care about the prompt but it throws if null
+ PasswordCallback passwordCb = new PasswordCallback("prompt", false);
+ // TODO: should not get pwd as a String but as a char array...
+ int passwordLen = response.length - authcidNullPosition - 1;
+ String pwd = new String(response, authcidNullPosition + 1, passwordLen, "utf8");
+ passwordCb.setPassword(pwd.toCharArray());
+ AuthorizeCallback authzCb = new AuthorizeCallback(authzid, authzid);
+ Callback[] callbacks = new Callback[]{nameCb, passwordCb, authzCb};
+ _cbh.handle(callbacks);
+ _complete = true;
+ if (authzCb.isAuthorized())
+ {
+ _authorizationId = authzCb.getAuthenticationID();
+ return null;
+ }
+ else
+ {
+ throw new SaslException("Authentication failed");
+ }
+ }
+ catch (IOException e)
+ {
+ throw new SaslException("Error processing data: " + e, e);
+ }
+ catch (UnsupportedCallbackException e)
+ {
+ throw new SaslException("Unable to obtain data from callback handler: " + e, e);
+ }
+ }
+
+ private int findNullPosition(byte[] response, int startPosition)
+ {
+ int position = startPosition;
+ while (position < response.length)
+ {
+ if (response[position] == (byte) 0)
+ {
+ return position;
+ }
+ position++;
+ }
+ return -1;
+ }
+
+ public boolean isComplete()
+ {
+ return _complete;
+ }
+
+ public String getAuthorizationID()
+ {
+ return _authorizationId;
+ }
+
+ public byte[] unwrap(byte[] incoming, int offset, int len) throws SaslException
+ {
+ throw new SaslException("Unsupported operation");
+ }
+
+ public byte[] wrap(byte[] outgoing, int offset, int len) throws SaslException
+ {
+ throw new SaslException("Unsupported operation");
+ }
+
+ public Object getNegotiatedProperty(String propName)
+ {
+ return null;
+ }
+
+ public void dispose() throws SaslException
+ {
+ _cbh = null;
+ }
+
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainSaslServer.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainSaslServerFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainSaslServerFactory.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainSaslServerFactory.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainSaslServerFactory.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,56 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.security.auth.plain;
+
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import javax.security.sasl.SaslServerFactory;
+import java.util.Map;
+
+public class PlainSaslServerFactory implements SaslServerFactory
+{
+ public SaslServer createSaslServer(String mechanism, String protocol, String serverName, Map props,
+ CallbackHandler cbh) throws SaslException
+ {
+ if (PlainSaslServer.MECHANISM.equals(mechanism))
+ {
+ return new PlainSaslServer(cbh);
+ }
+ else
+ {
+ return null;
+ }
+ }
+
+ public String[] getMechanismNames(Map props)
+ {
+ if (props.containsKey(Sasl.POLICY_NOPLAINTEXT) ||
+ props.containsKey(Sasl.POLICY_NODICTIONARY) ||
+ props.containsKey(Sasl.POLICY_NOACTIVE))
+ {
+ // returned array must be non null according to interface documentation
+ return new String[0];
+ }
+ else
+ {
+ return new String[]{PlainSaslServer.MECHANISM};
+ }
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/security/auth/plain/PlainSaslServerFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/AMQState.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/AMQState.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/AMQState.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/AMQState.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,33 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.state;
+
+/**
+ * States used in the AMQ protocol. Used by the finite state machine to determine
+ * valid responses.
+ */
+public enum AMQState
+{
+ CONNECTION_NOT_STARTED,
+ CONNECTION_NOT_AUTH,
+ CONNECTION_NOT_TUNED,
+ CONNECTION_NOT_OPENED,
+ CONNECTION_OPEN,
+ CONNECTION_CLOSING,
+ CONNECTION_CLOSED
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/AMQState.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/AMQStateManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/AMQStateManager.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/AMQStateManager.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/AMQStateManager.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,219 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.state;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.*;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.handler.*;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.protocol.AMQMethodListener;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.log4j.Logger;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArraySet;
+
+/**
+ * The state manager is responsible for managing the state of the protocol session.
+ * <p/>
+ * For each AMQProtocolHandler there is a separate state manager.
+ *
+ */
+public class AMQStateManager implements AMQMethodListener
+{
+ private static final Logger _logger = Logger.getLogger(AMQStateManager.class);
+
+ /**
+ * The current state
+ */
+ private AMQState _currentState;
+
+ /**
+ * Maps from an AMQState instance to a Map from Class to StateTransitionHandler.
+ * The class must be a subclass of AMQFrame.
+ */
+ private final Map<AMQState, Map<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>> _state2HandlersMap =
+ new HashMap<AMQState, Map<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>>();
+
+ private CopyOnWriteArraySet<StateListener> _stateListeners = new CopyOnWriteArraySet<StateListener>();
+
+ public AMQStateManager()
+ {
+ this(AMQState.CONNECTION_NOT_STARTED, true);
+ }
+
+ protected AMQStateManager(AMQState initial, boolean register)
+ {
+ _currentState = initial;
+ if (register)
+ {
+ registerListeners();
+ }
+ }
+
+ protected void registerListeners()
+ {
+ Map<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>> frame2handlerMap =
+ new HashMap<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>();
+
+ // we need to register a map for the null (i.e. all state) handlers otherwise you get
+ // a stack overflow in the handler searching code when you present it with a frame for which
+ // no handlers are registered
+ //
+ _state2HandlersMap.put(null, frame2handlerMap);
+
+ frame2handlerMap = new HashMap<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>();
+ frame2handlerMap.put(ConnectionStartOkBody.class, ConnectionStartOkMethodHandler.getInstance());
+ _state2HandlersMap.put(AMQState.CONNECTION_NOT_STARTED, frame2handlerMap);
+
+ frame2handlerMap = new HashMap<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>();
+ frame2handlerMap.put(ConnectionSecureOkBody.class, ConnectionSecureOkMethodHandler.getInstance());
+ _state2HandlersMap.put(AMQState.CONNECTION_NOT_AUTH, frame2handlerMap);
+
+ frame2handlerMap = new HashMap<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>();
+ frame2handlerMap.put(ConnectionTuneOkBody.class, ConnectionTuneOkMethodHandler.getInstance());
+ _state2HandlersMap.put(AMQState.CONNECTION_NOT_TUNED, frame2handlerMap);
+
+ frame2handlerMap = new HashMap<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>();
+ frame2handlerMap.put(ConnectionOpenBody.class, ConnectionOpenMethodHandler.getInstance());
+ _state2HandlersMap.put(AMQState.CONNECTION_NOT_OPENED, frame2handlerMap);
+
+ //
+ // ConnectionOpen handlers
+ //
+ frame2handlerMap = new HashMap<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>();
+ frame2handlerMap.put(ChannelOpenBody.class, ChannelOpenHandler.getInstance());
+ frame2handlerMap.put(ChannelCloseBody.class, ChannelCloseHandler.getInstance());
+ frame2handlerMap.put(ChannelCloseOkBody.class, ChannelCloseOkHandler.getInstance());
+ frame2handlerMap.put(ConnectionCloseBody.class, ConnectionCloseMethodHandler.getInstance());
+ frame2handlerMap.put(ExchangeDeclareBody.class, ExchangeDeclareHandler.getInstance());
+ frame2handlerMap.put(ExchangeDeleteBody.class, ExchangeDeleteHandler.getInstance());
+ frame2handlerMap.put(BasicAckBody.class, BasicAckMethodHandler.getInstance());
+ frame2handlerMap.put(BasicRecoverBody.class, BasicRecoverMethodHandler.getInstance());
+ frame2handlerMap.put(BasicConsumeBody.class, BasicConsumeMethodHandler.getInstance());
+ frame2handlerMap.put(BasicCancelBody.class, BasicCancelMethodHandler.getInstance());
+ frame2handlerMap.put(BasicPublishBody.class, BasicPublishMethodHandler.getInstance());
+ frame2handlerMap.put(BasicQosBody.class, BasicQosHandler.getInstance());
+ frame2handlerMap.put(QueueBindBody.class, QueueBindHandler.getInstance());
+ frame2handlerMap.put(QueueDeclareBody.class, QueueDeclareHandler.getInstance());
+ frame2handlerMap.put(QueueDeleteBody.class, QueueDeleteHandler.getInstance());
+ frame2handlerMap.put(ChannelFlowBody.class, ChannelFlowHandler.getInstance());
+ frame2handlerMap.put(TxSelectBody.class, TxSelectHandler.getInstance());
+ frame2handlerMap.put(TxCommitBody.class, TxCommitHandler.getInstance());
+ frame2handlerMap.put(TxRollbackBody.class, TxRollbackHandler.getInstance());
+
+ _state2HandlersMap.put(AMQState.CONNECTION_OPEN, frame2handlerMap);
+
+ frame2handlerMap = new HashMap<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>();
+ frame2handlerMap.put(ConnectionCloseOkBody.class, ConnectionCloseOkMethodHandler.getInstance());
+ _state2HandlersMap.put(AMQState.CONNECTION_CLOSING, frame2handlerMap);
+
+ }
+
+ public AMQState getCurrentState()
+ {
+ return _currentState;
+ }
+
+ public void changeState(AMQState newState) throws AMQException
+ {
+ _logger.debug("State changing to " + newState + " from old state " + _currentState);
+ final AMQState oldState = _currentState;
+ _currentState = newState;
+
+ for (StateListener l : _stateListeners)
+ {
+ l.stateChanged(oldState, newState);
+ }
+ }
+
+ public void error(AMQException e)
+ {
+ _logger.error("State manager received error notification: " + e, e);
+ for (StateListener l : _stateListeners)
+ {
+ l.error(e);
+ }
+ }
+
+ public <B extends AMQMethodBody> boolean methodReceived(AMQMethodEvent<B> evt,
+ AMQProtocolSession protocolSession,
+ QueueRegistry queueRegistry,
+ ExchangeRegistry exchangeRegistry) throws AMQException
+ {
+ StateAwareMethodListener<B> handler = findStateTransitionHandler(_currentState, evt.getMethod());
+ if (handler != null)
+ {
+ handler.methodReceived(this, queueRegistry, exchangeRegistry, protocolSession, evt);
+ return true;
+ }
+ return false;
+ }
+
+ protected <B extends AMQMethodBody> StateAwareMethodListener<B> findStateTransitionHandler(AMQState currentState,
+ B frame)
+ throws IllegalStateTransitionException
+ {
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Looking for state transition handler for frame " + frame.getClass());
+ }
+ final Map<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>
+ classToHandlerMap = _state2HandlersMap.get(currentState);
+
+ if (classToHandlerMap == null)
+ {
+ // if no specialised per state handler is registered look for a
+ // handler registered for "all" states
+ return findStateTransitionHandler(null, frame);
+ }
+ final StateAwareMethodListener<B> handler = (StateAwareMethodListener<B>) classToHandlerMap.get(frame.getClass());
+ if (handler == null)
+ {
+ if (currentState == null)
+ {
+ _logger.debug("No state transition handler defined for receiving frame " + frame);
+ return null;
+ }
+ else
+ {
+ // if no specialised per state handler is registered look for a
+ // handler registered for "all" states
+ return findStateTransitionHandler(null, frame);
+ }
+ }
+ else
+ {
+ return handler;
+ }
+ }
+
+ public void addStateListener(StateListener listener)
+ {
+ _logger.debug("Adding state listener");
+ _stateListeners.add(listener);
+ }
+
+ public void removeStateListener(StateListener listener)
+ {
+ _stateListeners.remove(listener);
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/AMQStateManager.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/IllegalStateTransitionException.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/IllegalStateTransitionException.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/IllegalStateTransitionException.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/IllegalStateTransitionException.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,45 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.state;
+
+import org.apache.qpid.AMQException;
+
+public class IllegalStateTransitionException extends AMQException
+{
+ private AMQState _originalState;
+
+ private Class _frame;
+
+ public IllegalStateTransitionException(AMQState originalState, Class frame)
+ {
+ super("No valid state transition defined for receiving frame " + frame +
+ " from state " + originalState);
+ _originalState = originalState;
+ _frame = frame;
+ }
+
+ public AMQState getOriginalState()
+ {
+ return _originalState;
+ }
+
+ public Class getFrameClass()
+ {
+ return _frame;
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/IllegalStateTransitionException.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/StateAwareMethodListener.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/StateAwareMethodListener.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/StateAwareMethodListener.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/StateAwareMethodListener.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,37 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.state;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.framing.AMQMethodBody;
+
+/**
+ * A frame listener that is informed of the protocol state when invoked and has
+ * the opportunity to update state.
+ *
+ */
+public interface StateAwareMethodListener <B extends AMQMethodBody>
+{
+ void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
+ ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
+ AMQMethodEvent<B> evt) throws AMQException;
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/StateAwareMethodListener.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/StateListener.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/StateListener.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/StateListener.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/StateListener.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,27 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.state;
+
+import org.apache.qpid.AMQException;
+
+public interface StateListener
+{
+ void stateChanged(AMQState oldState, AMQState newState) throws AMQException;
+
+ void error(Throwable t);
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/state/StateListener.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/store/MemoryMessageStore.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/store/MemoryMessageStore.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/store/MemoryMessageStore.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/store/MemoryMessageStore.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,137 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.store;
+
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.AMQException;
+import org.apache.log4j.Logger;
+import org.apache.commons.configuration.Configuration;
+
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.List;
+
+/**
+ * A simple message store that stores the messages in a threadsafe structure in memory.
+ *
+ */
+public class MemoryMessageStore implements MessageStore
+{
+ private static final Logger _log = Logger.getLogger(MemoryMessageStore.class);
+
+ private static final int DEFAULT_HASHTABLE_CAPACITY = 50000;
+
+ private static final String HASHTABLE_CAPACITY_CONFIG = "hashtable-capacity";
+
+ protected ConcurrentMap<Long, AMQMessage> _messageMap;
+
+ private final AtomicLong _messageId = new AtomicLong(1);
+
+ public void configure(String base, Configuration config)
+ {
+ int hashtableCapacity = config.getInt(base + "." + HASHTABLE_CAPACITY_CONFIG, DEFAULT_HASHTABLE_CAPACITY);
+ _log.info("Using capacity " + hashtableCapacity + " for hash table");
+ _messageMap = new ConcurrentHashMap<Long, AMQMessage>(hashtableCapacity);
+ }
+
+ public void configure(QueueRegistry queueRegistry, String base, Configuration config) throws Exception
+ {
+ configure(base, config);
+ }
+
+ public void close() throws Exception
+ {
+ if(_messageMap != null)
+ {
+ _messageMap.clear();
+ _messageMap = null;
+ }
+ }
+
+ public void put(AMQMessage msg)
+ {
+ _messageMap.put(msg.getMessageId(), msg);
+ }
+
+ public void removeMessage(long messageId)
+ {
+ if (_log.isDebugEnabled())
+ {
+ _log.debug("Removing message with id " + messageId);
+ }
+ _messageMap.remove(messageId);
+ }
+
+ public void createQueue(AMQQueue queue) throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void removeQueue(String name) throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void enqueueMessage(String name, long messageId) throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void dequeueMessage(String name, long messageId) throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void beginTran() throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void commitTran() throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void abortTran() throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public boolean inTran()
+ {
+ return false;
+ }
+
+ public List<AMQQueue> createQueues() throws AMQException
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public long getNewMessageId()
+ {
+ return _messageId.getAndIncrement();
+ }
+
+ public AMQMessage getMessage(long messageId)
+ {
+ return _messageMap.get(messageId);
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/store/MemoryMessageStore.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/store/MessageStore.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/store/MessageStore.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/store/MessageStore.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/store/MessageStore.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,80 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.store;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.QueueRegistry;
+
+import java.util.List;
+
+public interface MessageStore
+{
+ /**
+ * Called after instantiation in order to configure the message store. A particular implementation can define
+ * whatever parameters it wants.
+ * @param queueRegistry the registry of queues to be used by this store
+ * @param base the base element identifier from which all configuration items are relative. For example, if the base
+ * element is "store", the all elements used by concrete classes will be "store.foo" etc.
+ * @param config the apache commons configuration object
+ */
+ void configure(QueueRegistry queueRegistry, String base, Configuration config) throws Exception;
+
+ /**
+ * Called to close and cleanup any resources used by the message store.
+ * @throws Exception
+ */
+ void close() throws Exception;
+
+ void put(AMQMessage msg) throws AMQException;
+
+ void removeMessage(long messageId) throws AMQException;
+
+ void createQueue(AMQQueue queue) throws AMQException;
+
+ void removeQueue(String name) throws AMQException;
+
+ void enqueueMessage(String name, long messageId) throws AMQException;
+
+ void dequeueMessage(String name, long messageId) throws AMQException;
+
+ void beginTran() throws AMQException;
+
+ void commitTran() throws AMQException;
+
+ void abortTran() throws AMQException;
+
+ boolean inTran();
+
+ /**
+ * Recreate all queues that were persisted, including re-enqueuing of existing messages
+ * @return
+ * @throws AMQException
+ */
+ List<AMQQueue> createQueues() throws AMQException;
+
+ /**
+ * Return a valid, currently unused message id.
+ * @return a message id
+ */
+ long getNewMessageId();
+}
+
+
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/store/MessageStore.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/transport/ConnectorConfiguration.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/transport/ConnectorConfiguration.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/transport/ConnectorConfiguration.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/transport/ConnectorConfiguration.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,93 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.transport;
+
+import org.apache.qpid.configuration.Configured;
+import org.apache.mina.common.IoAcceptor;
+
+public class ConnectorConfiguration
+{
+ public static final String DEFAULT_PORT = "5672";
+
+ public static final String SSL_PORT = "8672";
+
+ @Configured(path = "connector.processors",
+ defaultValue = "4")
+ public int processors;
+
+ @Configured(path = "connector.port",
+ defaultValue = DEFAULT_PORT)
+ public int port;
+
+ @Configured(path = "connector.bind",
+ defaultValue = "wildcard")
+ public String bindAddress;
+
+ @Configured(path = "connector.sslport",
+ defaultValue = SSL_PORT)
+ public int sslPort;
+
+ @Configured(path = "connector.socketReceiveBuffer",
+ defaultValue = "32767")
+ public int socketReceiveBufferSize;
+
+ @Configured(path = "connector.socketWriteBuffer",
+ defaultValue = "32767")
+ public int socketWriteBuferSize;
+
+ @Configured(path = "connector.tcpNoDelay",
+ defaultValue = "true")
+ public boolean tcpNoDelay;
+
+ @Configured(path = "advanced.filterchain[@enableExecutorPool]",
+ defaultValue = "false")
+ public boolean enableExecutorPool;
+
+ @Configured(path = "advanced.enablePooledAllocator",
+ defaultValue = "false")
+ public boolean enablePooledAllocator;
+
+ @Configured(path = "advanced.enableDirectBuffers",
+ defaultValue = "false")
+ public boolean enableDirectBuffers;
+
+ @Configured(path = "connector.ssl",
+ defaultValue = "false")
+ public boolean enableSSL;
+
+ @Configured(path = "connector.nonssl",
+ defaultValue = "true")
+ public boolean enableNonSSL;
+
+ @Configured(path = "advanced.useBlockingIo",
+ defaultValue = "false")
+ public boolean useBlockingIo;
+
+ public IoAcceptor createAcceptor()
+ {
+ if(useBlockingIo)
+ {
+ System.out.println("Using blocking io");
+ return new org.apache.qpid.bio.SocketAcceptor();
+ }
+ else
+ {
+ return new org.apache.mina.transport.socket.nio.SocketAcceptor(processors);
+ }
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/transport/ConnectorConfiguration.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/transport/ThreadPoolFilter.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/transport/ThreadPoolFilter.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/transport/ThreadPoolFilter.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/transport/ThreadPoolFilter.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,692 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.transport;
+
+import org.apache.mina.common.*;
+import org.apache.mina.util.*;
+import org.apache.mina.util.Queue;
+import org.apache.mina.util.Stack;
+
+import java.util.*;
+
+/**
+ * A Thread-pooling filter. This filter forwards {@link IoHandler} events
+ * to its thread pool.
+ * <p/>
+ * This is an implementation of
+ * <a href="http://deuce.doc.wustl.edu/doc/pspdfs/lf.pdf">Leader/Followers
+ * thread pool</a> by Douglas C. Schmidt et al.
+ */
+public class ThreadPoolFilter extends IoFilterAdapter
+{
+ /**
+ * Default maximum size of thread pool (2G).
+ */
+ public static final int DEFAULT_MAXIMUM_POOL_SIZE = Integer.MAX_VALUE;
+
+ /**
+ * Default keep-alive time of thread pool (1 min).
+ */
+ public static final int DEFAULT_KEEP_ALIVE_TIME = 60 * 1000;
+
+ /**
+ * A queue which contains {@link Integer}s which represents reusable
+ * thread IDs. {@link Worker} first checks this queue and then
+ * uses {@link #threadId} when no reusable thread ID is available.
+ */
+ private static final Queue threadIdReuseQueue = new Queue();
+ private static int threadId = 0;
+
+ private static int acquireThreadId()
+ {
+ synchronized (threadIdReuseQueue)
+ {
+ Integer id = (Integer) threadIdReuseQueue.pop();
+ if (id == null)
+ {
+ return ++ threadId;
+ }
+ else
+ {
+ return id.intValue();
+ }
+ }
+ }
+
+ private static void releaseThreadId(int id)
+ {
+ synchronized (threadIdReuseQueue)
+ {
+ threadIdReuseQueue.push(new Integer(id));
+ }
+ }
+
+ private final String threadNamePrefix;
+ private final Map buffers = new IdentityHashMap();
+ private final BlockingQueue unfetchedSessionBuffers = new BlockingQueue();
+ private final Set allSessionBuffers = new IdentityHashSet();
+
+ private Worker leader;
+ private final Stack followers = new Stack();
+ private final Set allWorkers = new IdentityHashSet();
+
+ private int maximumPoolSize = DEFAULT_MAXIMUM_POOL_SIZE;
+ private int keepAliveTime = DEFAULT_KEEP_ALIVE_TIME;
+
+ private boolean shuttingDown;
+
+ private int poolSize;
+ private final Object poolSizeLock = new Object();
+
+ /**
+ * Creates a new instance of this filter with default thread pool settings.
+ */
+ public ThreadPoolFilter()
+ {
+ this("IoThreadPool");
+ }
+
+ /**
+ * Creates a new instance of this filter with the specified thread name prefix
+ * and other default settings.
+ *
+ * @param threadNamePrefix the prefix of the thread names this pool will create.
+ */
+ public ThreadPoolFilter(String threadNamePrefix)
+ {
+ if (threadNamePrefix == null)
+ {
+ throw new NullPointerException("threadNamePrefix");
+ }
+ threadNamePrefix = threadNamePrefix.trim();
+ if (threadNamePrefix.length() == 0)
+ {
+ throw new IllegalArgumentException("threadNamePrefix is empty.");
+ }
+ this.threadNamePrefix = threadNamePrefix;
+ }
+
+ public String getThreadNamePrefix()
+ {
+ return threadNamePrefix;
+ }
+
+ public int getPoolSize()
+ {
+ synchronized (poolSizeLock)
+ {
+ return poolSize;
+ }
+ }
+
+ public int getMaximumPoolSize()
+ {
+ return maximumPoolSize;
+ }
+
+ public int getKeepAliveTime()
+ {
+ return keepAliveTime;
+ }
+
+ public void setMaximumPoolSize(int maximumPoolSize)
+ {
+ if (maximumPoolSize <= 0)
+ {
+ throw new IllegalArgumentException();
+ }
+ this.maximumPoolSize = maximumPoolSize;
+ }
+
+ public void setKeepAliveTime(int keepAliveTime)
+ {
+ this.keepAliveTime = keepAliveTime;
+ }
+
+ public void init()
+ {
+ shuttingDown = false;
+ leader = new Worker();
+ leader.start();
+ leader.lead();
+ }
+
+ public void destroy()
+ {
+ shuttingDown = true;
+ int expectedPoolSize = 0;
+ while (getPoolSize() != expectedPoolSize)
+ {
+ List allWorkers;
+ synchronized (poolSizeLock)
+ {
+ allWorkers = new ArrayList(this.allWorkers);
+ }
+
+ // You may not interrupt the current thread.
+ if (allWorkers.remove(Thread.currentThread()))
+ {
+ expectedPoolSize = 1;
+ }
+
+ for (Iterator i = allWorkers.iterator(); i.hasNext();)
+ {
+ Worker worker = (Worker) i.next();
+ while (worker.isAlive())
+ {
+ worker.interrupt();
+ try
+ {
+ // This timeout will help us from
+ // infinite lock-up and interrupt workers again.
+ worker.join(100);
+ }
+ catch (InterruptedException e)
+ {
+ }
+ }
+ }
+ }
+
+ this.allSessionBuffers.clear();
+ this.unfetchedSessionBuffers.clear();
+ this.buffers.clear();
+ this.followers.clear();
+ this.leader = null;
+ }
+
+ private void increasePoolSize(Worker worker)
+ {
+ synchronized (poolSizeLock)
+ {
+ poolSize++;
+ allWorkers.add(worker);
+ }
+ }
+
+ private void decreasePoolSize(Worker worker)
+ {
+ synchronized (poolSizeLock)
+ {
+ poolSize--;
+ allWorkers.remove(worker);
+ }
+ }
+
+ private void fireEvent(NextFilter nextFilter, IoSession session,
+ EventType type, Object data)
+ {
+ final BlockingQueue unfetchedSessionBuffers = this.unfetchedSessionBuffers;
+ final Set allSessionBuffers = this.allSessionBuffers;
+ final Event event = new Event(type, nextFilter, data);
+
+ synchronized (unfetchedSessionBuffers)
+ {
+ final SessionBuffer buf = getSessionBuffer(session);
+ final Queue eventQueue = buf.eventQueue;
+
+ synchronized (buf)
+ {
+ eventQueue.push(event);
+ }
+
+ if (!allSessionBuffers.contains(buf))
+ {
+ allSessionBuffers.add(buf);
+ unfetchedSessionBuffers.push(buf);
+ }
+ }
+ }
+
+ /**
+ * Implement this method to fetch (or pop) a {@link SessionBuffer} from
+ * the given <tt>unfetchedSessionBuffers</tt>. The default implementation
+ * simply pops the buffer from it. You could prioritize the fetch order.
+ *
+ * @return A non-null {@link SessionBuffer}
+ */
+ protected SessionBuffer fetchSessionBuffer(Queue unfetchedSessionBuffers)
+ {
+ return (SessionBuffer) unfetchedSessionBuffers.pop();
+ }
+
+ private SessionBuffer getSessionBuffer(IoSession session)
+ {
+ final Map buffers = this.buffers;
+ SessionBuffer buf = (SessionBuffer) buffers.get(session);
+ if (buf == null)
+ {
+ synchronized (buffers)
+ {
+ buf = (SessionBuffer) buffers.get(session);
+ if (buf == null)
+ {
+ buf = new SessionBuffer(session);
+ buffers.put(session, buf);
+ }
+ }
+ }
+ return buf;
+ }
+
+ private void removeSessionBuffer(SessionBuffer buf)
+ {
+ final Map buffers = this.buffers;
+ final IoSession session = buf.session;
+ synchronized (buffers)
+ {
+ buffers.remove(session);
+ }
+ }
+
+ protected static class SessionBuffer
+ {
+ private final IoSession session;
+
+ private final Queue eventQueue = new Queue();
+
+ private SessionBuffer(IoSession session)
+ {
+ this.session = session;
+ }
+
+ public IoSession getSession()
+ {
+ return session;
+ }
+
+ public Queue getEventQueue()
+ {
+ return eventQueue;
+ }
+ }
+
+ private class Worker extends Thread
+ {
+ private final int id;
+ private final Object promotionLock = new Object();
+ private boolean dead;
+
+ private Worker()
+ {
+ int id = acquireThreadId();
+ this.id = id;
+ this.setName(threadNamePrefix + '-' + id);
+ increasePoolSize(this);
+ }
+
+ public boolean lead()
+ {
+ final Object promotionLock = this.promotionLock;
+ synchronized (promotionLock)
+ {
+ if (dead)
+ {
+ return false;
+ }
+
+ leader = this;
+ promotionLock.notify();
+ }
+
+ return true;
+ }
+
+ public void run()
+ {
+ for (; ;)
+ {
+ if (!waitForPromotion())
+ {
+ break;
+ }
+
+ SessionBuffer buf = fetchBuffer();
+ giveUpLead();
+ if (buf == null)
+ {
+ break;
+ }
+
+ processEvents(buf);
+ follow();
+ releaseBuffer(buf);
+ }
+
+ decreasePoolSize(this);
+ releaseThreadId(id);
+ }
+
+ private SessionBuffer fetchBuffer()
+ {
+ BlockingQueue unfetchedSessionBuffers = ThreadPoolFilter.this.unfetchedSessionBuffers;
+ synchronized (unfetchedSessionBuffers)
+ {
+ while (!shuttingDown)
+ {
+ try
+ {
+ unfetchedSessionBuffers.waitForNewItem();
+ }
+ catch (InterruptedException e)
+ {
+ continue;
+ }
+
+ return ThreadPoolFilter.this.fetchSessionBuffer(unfetchedSessionBuffers);
+ }
+ }
+
+ return null;
+ }
+
+ private void processEvents(SessionBuffer buf)
+ {
+ final IoSession session = buf.session;
+ final Queue eventQueue = buf.eventQueue;
+ for (; ;)
+ {
+ Event event;
+ synchronized (buf)
+ {
+ event = (Event) eventQueue.pop();
+ if (event == null)
+ {
+ break;
+ }
+ }
+ processEvent(event.getNextFilter(), session,
+ event.getType(), event.getData());
+ }
+ }
+
+ private void follow()
+ {
+ final Object promotionLock = this.promotionLock;
+ final Stack followers = ThreadPoolFilter.this.followers;
+ synchronized (promotionLock)
+ {
+ if (this != leader)
+ {
+ synchronized (followers)
+ {
+ followers.push(this);
+ }
+ }
+ }
+ }
+
+ private void releaseBuffer(SessionBuffer buf)
+ {
+ final BlockingQueue unfetchedSessionBuffers = ThreadPoolFilter.this.unfetchedSessionBuffers;
+ final Set allSessionBuffers = ThreadPoolFilter.this.allSessionBuffers;
+ final Queue eventQueue = buf.eventQueue;
+
+ synchronized (unfetchedSessionBuffers)
+ {
+ if (eventQueue.isEmpty())
+ {
+ allSessionBuffers.remove(buf);
+ removeSessionBuffer(buf);
+ }
+ else
+ {
+ unfetchedSessionBuffers.push(buf);
+ }
+ }
+ }
+
+ private boolean waitForPromotion()
+ {
+ final Object promotionLock = this.promotionLock;
+
+ long startTime = System.currentTimeMillis();
+ long currentTime = System.currentTimeMillis();
+
+ synchronized (promotionLock)
+ {
+ while (this != leader && !shuttingDown)
+ {
+ // Calculate remaining keep-alive time
+ int keepAliveTime = getKeepAliveTime();
+ if (keepAliveTime > 0)
+ {
+ keepAliveTime -= (currentTime - startTime);
+ }
+ else
+ {
+ keepAliveTime = Integer.MAX_VALUE;
+ }
+
+ // Break the loop if there's no remaining keep-alive time.
+ if (keepAliveTime <= 0)
+ {
+ break;
+ }
+
+ // Wait for promotion
+ try
+ {
+ promotionLock.wait(keepAliveTime);
+ }
+ catch (InterruptedException e)
+ {
+ }
+
+ // Update currentTime for the next iteration
+ currentTime = System.currentTimeMillis();
+ }
+
+ boolean timeToLead = this == leader && !shuttingDown;
+
+ if (!timeToLead)
+ {
+ // time to die
+ synchronized (followers)
+ {
+ followers.remove(this);
+ }
+
+ // Mark as dead explicitly when we've got promotionLock.
+ dead = true;
+ }
+
+ return timeToLead;
+ }
+ }
+
+ private void giveUpLead()
+ {
+ final Stack followers = ThreadPoolFilter.this.followers;
+ Worker worker;
+ do
+ {
+ synchronized (followers)
+ {
+ worker = (Worker) followers.pop();
+ }
+
+ if (worker == null)
+ {
+ // Increase the number of threads if we
+ // are not shutting down and we can increase the number.
+ if (!shuttingDown
+ && getPoolSize() < getMaximumPoolSize())
+ {
+ worker = new Worker();
+ worker.lead();
+ worker.start();
+ }
+
+ // This loop should end because:
+ // 1) lead() is called already,
+ // 2) or it is shutting down and there's no more threads left.
+ break;
+ }
+ }
+ while (!worker.lead());
+ }
+ }
+
+ protected static class EventType
+ {
+ public static final EventType OPENED = new EventType("OPENED");
+
+ public static final EventType CLOSED = new EventType("CLOSED");
+
+ public static final EventType READ = new EventType("READ");
+
+ public static final EventType WRITTEN = new EventType("WRITTEN");
+
+ public static final EventType RECEIVED = new EventType("RECEIVED");
+
+ public static final EventType SENT = new EventType("SENT");
+
+ public static final EventType IDLE = new EventType("IDLE");
+
+ public static final EventType EXCEPTION = new EventType("EXCEPTION");
+
+ private final String value;
+
+ private EventType(String value)
+ {
+ this.value = value;
+ }
+
+ public String toString()
+ {
+ return value;
+ }
+ }
+
+ protected static class Event
+ {
+ private final EventType type;
+ private final NextFilter nextFilter;
+ private final Object data;
+
+ public Event(EventType type, NextFilter nextFilter, Object data)
+ {
+ this.type = type;
+ this.nextFilter = nextFilter;
+ this.data = data;
+ }
+
+ public Object getData()
+ {
+ return data;
+ }
+
+
+ public NextFilter getNextFilter()
+ {
+ return nextFilter;
+ }
+
+
+ public EventType getType()
+ {
+ return type;
+ }
+ }
+
+ public void sessionCreated(NextFilter nextFilter, IoSession session)
+ {
+ nextFilter.sessionCreated(session);
+ }
+
+ public void sessionOpened(NextFilter nextFilter,
+ IoSession session)
+ {
+ fireEvent(nextFilter, session, EventType.OPENED, null);
+ }
+
+ public void sessionClosed(NextFilter nextFilter,
+ IoSession session)
+ {
+ fireEvent(nextFilter, session, EventType.CLOSED, null);
+ }
+
+ public void sessionIdle(NextFilter nextFilter,
+ IoSession session, IdleStatus status)
+ {
+ fireEvent(nextFilter, session, EventType.IDLE, status);
+ }
+
+ public void exceptionCaught(NextFilter nextFilter,
+ IoSession session, Throwable cause)
+ {
+ fireEvent(nextFilter, session, EventType.EXCEPTION, cause);
+ }
+
+ public void messageReceived(NextFilter nextFilter,
+ IoSession session, Object message)
+ {
+ ByteBufferUtil.acquireIfPossible(message);
+ fireEvent(nextFilter, session, EventType.RECEIVED, message);
+ }
+
+ public void messageSent(NextFilter nextFilter,
+ IoSession session, Object message)
+ {
+ ByteBufferUtil.acquireIfPossible(message);
+ fireEvent(nextFilter, session, EventType.SENT, message);
+ }
+
+ protected void processEvent(NextFilter nextFilter,
+ IoSession session, EventType type,
+ Object data)
+ {
+ if (type == EventType.RECEIVED)
+ {
+ nextFilter.messageReceived(session, data);
+ ByteBufferUtil.releaseIfPossible(data);
+ }
+ else if (type == EventType.SENT)
+ {
+ nextFilter.messageSent(session, data);
+ ByteBufferUtil.releaseIfPossible(data);
+ }
+ else if (type == EventType.EXCEPTION)
+ {
+ nextFilter.exceptionCaught(session, (Throwable) data);
+ }
+ else if (type == EventType.IDLE)
+ {
+ nextFilter.sessionIdle(session, (IdleStatus) data);
+ }
+ else if (type == EventType.OPENED)
+ {
+ nextFilter.sessionOpened(session);
+ }
+ else if (type == EventType.CLOSED)
+ {
+ nextFilter.sessionClosed(session);
+ }
+ }
+
+ public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest)
+ {
+ nextFilter.filterWrite(session, writeRequest);
+ }
+
+ public void filterClose(NextFilter nextFilter, IoSession session) throws Exception
+ {
+ nextFilter.filterClose(session);
+ }
+}
\ No newline at end of file
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/transport/ThreadPoolFilter.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/txn/TxnBuffer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/txn/TxnBuffer.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/txn/TxnBuffer.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/txn/TxnBuffer.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,75 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.txn;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.server.store.MessageStore;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class TxnBuffer
+{
+ private final MessageStore _store;
+ private final List<TxnOp> _ops = new ArrayList<TxnOp>();
+
+ public TxnBuffer(MessageStore store)
+ {
+ _store = store;
+ }
+
+ public void commit() throws AMQException
+ {
+ _store.beginTran();
+ boolean failed = true;
+ try
+ {
+ for(TxnOp op : _ops)
+ {
+ op.commit();
+ }
+ _ops.clear();
+ failed = false;
+ }
+ finally
+ {
+ if(failed)
+ {
+ _store.abortTran();
+ }
+ else
+ {
+ _store.commitTran();
+ }
+ }
+ }
+
+ public void rollback() throws AMQException
+ {
+ for(TxnOp op : _ops)
+ {
+ op.rollback();
+ }
+ _ops.clear();
+ }
+
+ public void enlist(TxnOp op)
+ {
+ _ops.add(op);
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/txn/TxnBuffer.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/txn/TxnOp.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/txn/TxnOp.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/txn/TxnOp.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/txn/TxnOp.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,26 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.txn;
+
+import org.apache.qpid.AMQException;
+
+public interface TxnOp
+{
+ public void commit() throws AMQException;
+ public void rollback();
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/txn/TxnOp.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/util/CircularBuffer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/util/CircularBuffer.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/util/CircularBuffer.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/util/CircularBuffer.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,123 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.util;
+
+import java.util.Iterator;
+
+public class CircularBuffer implements Iterable
+{
+ private final Object[] _log;
+ private int _size;
+ private int _index;
+
+ public CircularBuffer(int size)
+ {
+ _log = new Object[size];
+ }
+
+ public void add(Object o)
+ {
+ _log[_index++] = o;
+ _size = Math.min(_size+1, _log.length);
+ if(_index >= _log.length)
+ {
+ _index = 0;
+ }
+ }
+
+ public Object get(int i)
+ {
+ if(i >= _log.length)
+ {
+ throw new ArrayIndexOutOfBoundsException(i);
+ }
+ return _log[index(i)];
+ }
+
+ public int size() {
+ return _size;
+ }
+
+ public Iterator iterator()
+ {
+ return new Iterator()
+ {
+ private int i = 0;
+
+ public boolean hasNext()
+ {
+ return i < _size;
+ }
+
+ public Object next()
+ {
+ return get(i++);
+ }
+
+ public void remove()
+ {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+
+ public String toString()
+ {
+ StringBuilder s = new StringBuilder();
+ boolean first = true;
+ for(Object o : this)
+ {
+ if(!first)
+ {
+ s.append(", ");
+ }
+ else
+ {
+ first = false;
+ }
+ s.append(o);
+ }
+ return s.toString();
+ }
+
+ public void dump()
+ {
+ for(Object o : this)
+ {
+ System.out.println(o);
+ }
+ }
+
+ int index(int i)
+ {
+ return _size == _log.length ? (_index + i) % _log.length : i;
+ }
+
+ public static void main(String[] artgv)
+ {
+ String[] items = new String[]{
+ "A","B","C","D","E","F","G","H","I","J","K"
+ };
+ CircularBuffer buffer = new CircularBuffer(5);
+ for(String s : items)
+ {
+ buffer.add(s);
+ System.out.println(buffer);
+ }
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/util/CircularBuffer.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/util/LoggingProxy.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/util/LoggingProxy.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/util/LoggingProxy.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/util/LoggingProxy.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,102 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.util;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.util.Arrays;
+
+/**
+ * Dynamic proxy that records invocations in a fixed size circular buffer,
+ * dumping details on hitting an exception.
+ * <p>
+ * Useful in debugging.
+ * <p>
+ */
+public class LoggingProxy implements InvocationHandler
+{
+ private final Object _target;
+ private final CircularBuffer _log;
+
+ public LoggingProxy(Object target, int size)
+ {
+ _target = target;
+ _log = new CircularBuffer(size);
+ }
+
+ public Object invoke(Object proxy, Method method, Object[] args) throws Throwable
+ {
+ try
+ {
+ entered(method, args);
+ Object result = method.invoke(_target, args);
+ returned(method, result);
+ return result;
+ }
+ catch(InvocationTargetException e)
+ {
+ dump();
+ throw e.getTargetException();
+ }
+ }
+
+ void dump()
+ {
+ _log.dump();
+ }
+
+ CircularBuffer getBuffer()
+ {
+ return _log;
+ }
+
+ private synchronized void entered(Method method, Object[] args)
+ {
+ if (args == null)
+ {
+ _log.add(Thread.currentThread() + ": " + method.getName() + "() entered");
+ }
+ else
+ {
+ _log.add(Thread.currentThread() + ": " + method.getName() + "(" + Arrays.toString(args) + ") entered");
+ }
+ }
+
+ private synchronized void returned(Method method, Object result)
+ {
+ if (method.getReturnType() == Void.TYPE)
+ {
+ _log.add(Thread.currentThread() + ": " + method.getName() + "() returned");
+ }
+ else
+ {
+ _log.add(Thread.currentThread() + ": " + method.getName() + "() returned " + result);
+ }
+ }
+
+ public Object getProxy(Class... c)
+ {
+ return Proxy.newProxyInstance(_target.getClass().getClassLoader(), c, this);
+ }
+
+ public int getBufferSize() {
+ return _log.size();
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/broker/src/org/apache/qpid/server/util/LoggingProxy.java
------------------------------------------------------------------------------
svn:eol-style = native