qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From r..@apache.org
Subject 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...
Date Tue, 19 Sep 2006 22:07:25 GMT
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



Mime
View raw message