From 1db5a8a2329ec064d1683294ee1a3d8d233de42d Mon Sep 17 00:00:00 2001 From: Stephen Vinoski Date: Sat, 18 Nov 2006 02:12:32 +0000 Subject: directory moves required for maven merge git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@476414 13f79535-47bb-0310-9956-ffa450edef68 --- .../org/apache/qpid/AMQChannelClosedException.java | 34 ++ .../java/org/apache/qpid/AMQChannelException.java | 49 ++ .../apache/qpid/AMQConnectionClosedException.java | 34 ++ .../org/apache/qpid/AMQConnectionException.java | 30 ++ .../org/apache/qpid/AMQDisconnectedException.java | 34 ++ .../main/java/org/apache/qpid/AMQException.java | 77 +++ .../org/apache/qpid/AMQUndeliveredException.java | 44 ++ .../apache/qpid/AMQUnresolvedAddressException.java | 37 ++ .../org/apache/qpid/codec/AMQCodecFactory.java | 53 ++ .../java/org/apache/qpid/codec/AMQDecoder.java | 99 ++++ .../java/org/apache/qpid/codec/AMQEncoder.java | 41 ++ .../org/apache/qpid/configuration/Configured.java | 44 ++ .../qpid/configuration/PropertyException.java | 65 +++ .../apache/qpid/configuration/PropertyUtils.java | 156 ++++++ .../org/apache/qpid/exchange/ExchangeDefaults.java | 36 ++ .../main/java/org/apache/qpid/framing/AMQBody.java | 38 ++ .../java/org/apache/qpid/framing/AMQDataBlock.java | 43 ++ .../apache/qpid/framing/AMQDataBlockDecoder.java | 116 ++++ .../apache/qpid/framing/AMQDataBlockEncoder.java | 65 +++ .../java/org/apache/qpid/framing/AMQFrame.java | 76 +++ .../qpid/framing/AMQFrameDecodingException.java | 48 ++ .../org/apache/qpid/framing/AMQMethodBody.java | 90 ++++ .../apache/qpid/framing/AMQMethodBodyFactory.java | 46 ++ .../qpid/framing/AMQProtocolClassException.java | 29 + .../qpid/framing/AMQProtocolHeaderException.java | 31 ++ .../qpid/framing/AMQProtocolInstanceException.java | 29 + .../qpid/framing/AMQProtocolVersionException.java | 33 ++ .../qpid/framing/BasicContentHeaderProperties.java | 592 +++++++++++++++++++++ .../java/org/apache/qpid/framing/BodyFactory.java | 31 ++ .../apache/qpid/framing/CompositeAMQDataBlock.java | 103 ++++ .../java/org/apache/qpid/framing/ContentBody.java | 85 +++ .../apache/qpid/framing/ContentBodyFactory.java | 47 ++ .../org/apache/qpid/framing/ContentHeaderBody.java | 115 ++++ .../qpid/framing/ContentHeaderBodyFactory.java | 50 ++ .../qpid/framing/ContentHeaderProperties.java | 58 ++ .../framing/ContentHeaderPropertiesFactory.java | 54 ++ .../apache/qpid/framing/EncodableAMQDataBlock.java | 35 ++ .../org/apache/qpid/framing/EncodingUtils.java | 549 +++++++++++++++++++ .../java/org/apache/qpid/framing/FieldTable.java | 322 +++++++++++ .../qpid/framing/FieldTableKeyEnumeration.java | 47 ++ .../org/apache/qpid/framing/HeartbeatBody.java | 57 ++ .../apache/qpid/framing/HeartbeatBodyFactory.java | 31 ++ .../apache/qpid/framing/ProtocolInitiation.java | 179 +++++++ .../src/main/java/org/apache/qpid/pool/Event.java | 114 ++++ .../src/main/java/org/apache/qpid/pool/Job.java | 113 ++++ .../java/org/apache/qpid/pool/PoolingFilter.java | 186 +++++++ .../org/apache/qpid/pool/ReadWriteThreadModel.java | 40 ++ .../pool/ReferenceCountingExecutorService.java | 98 ++++ .../java/org/apache/qpid/protocol/AMQConstant.java | 108 ++++ .../apache/qpid/ssl/BogusSSLContextFactory.java | 159 ++++++ .../apache/qpid/ssl/BogusTrustManagerFactory.java | 82 +++ .../apache/qpid/ssl/SSLServerSocketFactory.java | 108 ++++ .../java/org/apache/qpid/ssl/SSLSocketFactory.java | 138 +++++ .../java/org/apache/qpid/url/AMQBindingURL.java | 263 +++++++++ .../main/java/org/apache/qpid/url/BindingURL.java | 68 +++ .../main/java/org/apache/qpid/url/URLHelper.java | 176 ++++++ .../org/apache/qpid/url/URLSyntaxException.java | 97 ++++ .../qpid/util/ConcurrentLinkedQueueAtomicSize.java | 58 ++ .../qpid/util/ConcurrentLinkedQueueNoSize.java | 38 ++ .../src/main/resources/ProtocolVersionList.java | 40 ++ java/common/src/main/resources/cluster.asl | 59 ++ .../main/resources/org/apache/qpid/ssl/qpid.cert | Bin 0 -> 756 bytes java/common/src/main/resources/registry.template | 25 + .../org/apache/qpid/AMQChannelClosedException.java | 34 -- .../src/org/apache/qpid/AMQChannelException.java | 49 -- .../apache/qpid/AMQConnectionClosedException.java | 34 -- .../org/apache/qpid/AMQConnectionException.java | 30 -- .../org/apache/qpid/AMQDisconnectedException.java | 34 -- java/common/src/org/apache/qpid/AMQException.java | 77 --- .../org/apache/qpid/AMQUndeliveredException.java | 44 -- .../apache/qpid/AMQUnresolvedAddressException.java | 37 -- .../src/org/apache/qpid/codec/AMQCodecFactory.java | 53 -- .../src/org/apache/qpid/codec/AMQDecoder.java | 99 ---- .../src/org/apache/qpid/codec/AMQEncoder.java | 41 -- .../org/apache/qpid/configuration/Configured.java | 44 -- .../qpid/configuration/PropertyException.java | 65 --- .../apache/qpid/configuration/PropertyUtils.java | 156 ------ .../org/apache/qpid/exchange/ExchangeDefaults.java | 36 -- .../src/org/apache/qpid/framing/AMQBody.java | 38 -- .../src/org/apache/qpid/framing/AMQDataBlock.java | 43 -- .../apache/qpid/framing/AMQDataBlockDecoder.java | 116 ---- .../apache/qpid/framing/AMQDataBlockEncoder.java | 65 --- .../src/org/apache/qpid/framing/AMQFrame.java | 76 --- .../qpid/framing/AMQFrameDecodingException.java | 48 -- .../src/org/apache/qpid/framing/AMQMethodBody.java | 90 ---- .../apache/qpid/framing/AMQMethodBodyFactory.java | 46 -- .../qpid/framing/AMQProtocolClassException.java | 29 - .../qpid/framing/AMQProtocolHeaderException.java | 31 -- .../qpid/framing/AMQProtocolInstanceException.java | 29 - .../qpid/framing/AMQProtocolVersionException.java | 33 -- .../qpid/framing/BasicContentHeaderProperties.java | 592 --------------------- .../src/org/apache/qpid/framing/BodyFactory.java | 31 -- .../apache/qpid/framing/CompositeAMQDataBlock.java | 103 ---- .../src/org/apache/qpid/framing/ContentBody.java | 85 --- .../apache/qpid/framing/ContentBodyFactory.java | 47 -- .../org/apache/qpid/framing/ContentHeaderBody.java | 115 ---- .../qpid/framing/ContentHeaderBodyFactory.java | 50 -- .../qpid/framing/ContentHeaderProperties.java | 58 -- .../framing/ContentHeaderPropertiesFactory.java | 54 -- .../apache/qpid/framing/EncodableAMQDataBlock.java | 35 -- .../src/org/apache/qpid/framing/EncodingUtils.java | 549 ------------------- .../src/org/apache/qpid/framing/FieldTable.java | 322 ----------- .../qpid/framing/FieldTableKeyEnumeration.java | 47 -- .../src/org/apache/qpid/framing/HeartbeatBody.java | 57 -- .../apache/qpid/framing/HeartbeatBodyFactory.java | 31 -- .../apache/qpid/framing/ProtocolInitiation.java | 179 ------- java/common/src/org/apache/qpid/pool/Event.java | 114 ---- java/common/src/org/apache/qpid/pool/Job.java | 113 ---- .../src/org/apache/qpid/pool/PoolingFilter.java | 186 ------- .../org/apache/qpid/pool/ReadWriteThreadModel.java | 40 -- .../pool/ReferenceCountingExecutorService.java | 98 ---- .../src/org/apache/qpid/protocol/AMQConstant.java | 108 ---- .../apache/qpid/ssl/BogusSSLContextFactory.java | 159 ------ .../apache/qpid/ssl/BogusTrustManagerFactory.java | 82 --- .../apache/qpid/ssl/SSLServerSocketFactory.java | 108 ---- .../src/org/apache/qpid/ssl/SSLSocketFactory.java | 138 ----- .../src/org/apache/qpid/url/AMQBindingURL.java | 263 --------- .../common/src/org/apache/qpid/url/BindingURL.java | 68 --- java/common/src/org/apache/qpid/url/URLHelper.java | 176 ------ .../org/apache/qpid/url/URLSyntaxException.java | 97 ---- .../qpid/util/ConcurrentLinkedQueueAtomicSize.java | 58 -- .../qpid/util/ConcurrentLinkedQueueNoSize.java | 38 -- 122 files changed, 5802 insertions(+), 5678 deletions(-) create mode 100644 java/common/src/main/java/org/apache/qpid/AMQChannelClosedException.java create mode 100644 java/common/src/main/java/org/apache/qpid/AMQChannelException.java create mode 100644 java/common/src/main/java/org/apache/qpid/AMQConnectionClosedException.java create mode 100644 java/common/src/main/java/org/apache/qpid/AMQConnectionException.java create mode 100644 java/common/src/main/java/org/apache/qpid/AMQDisconnectedException.java create mode 100644 java/common/src/main/java/org/apache/qpid/AMQException.java create mode 100644 java/common/src/main/java/org/apache/qpid/AMQUndeliveredException.java create mode 100644 java/common/src/main/java/org/apache/qpid/AMQUnresolvedAddressException.java create mode 100644 java/common/src/main/java/org/apache/qpid/codec/AMQCodecFactory.java create mode 100644 java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java create mode 100644 java/common/src/main/java/org/apache/qpid/codec/AMQEncoder.java create mode 100644 java/common/src/main/java/org/apache/qpid/configuration/Configured.java create mode 100644 java/common/src/main/java/org/apache/qpid/configuration/PropertyException.java create mode 100644 java/common/src/main/java/org/apache/qpid/configuration/PropertyUtils.java create mode 100644 java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/AMQBody.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/AMQDataBlock.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockDecoder.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockEncoder.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/AMQFrame.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/AMQFrameDecodingException.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/AMQMethodBody.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/AMQMethodBodyFactory.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/AMQProtocolClassException.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/AMQProtocolHeaderException.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/AMQProtocolInstanceException.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/AMQProtocolVersionException.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/BodyFactory.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/ContentBody.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/ContentBodyFactory.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBodyFactory.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/ContentHeaderProperties.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/EncodableAMQDataBlock.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/FieldTable.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/FieldTableKeyEnumeration.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/HeartbeatBody.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/HeartbeatBodyFactory.java create mode 100644 java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java create mode 100644 java/common/src/main/java/org/apache/qpid/pool/Event.java create mode 100644 java/common/src/main/java/org/apache/qpid/pool/Job.java create mode 100644 java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java create mode 100644 java/common/src/main/java/org/apache/qpid/pool/ReadWriteThreadModel.java create mode 100644 java/common/src/main/java/org/apache/qpid/pool/ReferenceCountingExecutorService.java create mode 100644 java/common/src/main/java/org/apache/qpid/protocol/AMQConstant.java create mode 100644 java/common/src/main/java/org/apache/qpid/ssl/BogusSSLContextFactory.java create mode 100644 java/common/src/main/java/org/apache/qpid/ssl/BogusTrustManagerFactory.java create mode 100644 java/common/src/main/java/org/apache/qpid/ssl/SSLServerSocketFactory.java create mode 100644 java/common/src/main/java/org/apache/qpid/ssl/SSLSocketFactory.java create mode 100644 java/common/src/main/java/org/apache/qpid/url/AMQBindingURL.java create mode 100644 java/common/src/main/java/org/apache/qpid/url/BindingURL.java create mode 100644 java/common/src/main/java/org/apache/qpid/url/URLHelper.java create mode 100644 java/common/src/main/java/org/apache/qpid/url/URLSyntaxException.java create mode 100644 java/common/src/main/java/org/apache/qpid/util/ConcurrentLinkedQueueAtomicSize.java create mode 100644 java/common/src/main/java/org/apache/qpid/util/ConcurrentLinkedQueueNoSize.java create mode 100644 java/common/src/main/resources/ProtocolVersionList.java create mode 100644 java/common/src/main/resources/cluster.asl create mode 100644 java/common/src/main/resources/org/apache/qpid/ssl/qpid.cert create mode 100644 java/common/src/main/resources/registry.template delete mode 100644 java/common/src/org/apache/qpid/AMQChannelClosedException.java delete mode 100644 java/common/src/org/apache/qpid/AMQChannelException.java delete mode 100644 java/common/src/org/apache/qpid/AMQConnectionClosedException.java delete mode 100644 java/common/src/org/apache/qpid/AMQConnectionException.java delete mode 100644 java/common/src/org/apache/qpid/AMQDisconnectedException.java delete mode 100644 java/common/src/org/apache/qpid/AMQException.java delete mode 100644 java/common/src/org/apache/qpid/AMQUndeliveredException.java delete mode 100644 java/common/src/org/apache/qpid/AMQUnresolvedAddressException.java delete mode 100644 java/common/src/org/apache/qpid/codec/AMQCodecFactory.java delete mode 100644 java/common/src/org/apache/qpid/codec/AMQDecoder.java delete mode 100644 java/common/src/org/apache/qpid/codec/AMQEncoder.java delete mode 100644 java/common/src/org/apache/qpid/configuration/Configured.java delete mode 100644 java/common/src/org/apache/qpid/configuration/PropertyException.java delete mode 100644 java/common/src/org/apache/qpid/configuration/PropertyUtils.java delete mode 100644 java/common/src/org/apache/qpid/exchange/ExchangeDefaults.java delete mode 100644 java/common/src/org/apache/qpid/framing/AMQBody.java delete mode 100644 java/common/src/org/apache/qpid/framing/AMQDataBlock.java delete mode 100644 java/common/src/org/apache/qpid/framing/AMQDataBlockDecoder.java delete mode 100644 java/common/src/org/apache/qpid/framing/AMQDataBlockEncoder.java delete mode 100644 java/common/src/org/apache/qpid/framing/AMQFrame.java delete mode 100644 java/common/src/org/apache/qpid/framing/AMQFrameDecodingException.java delete mode 100644 java/common/src/org/apache/qpid/framing/AMQMethodBody.java delete mode 100644 java/common/src/org/apache/qpid/framing/AMQMethodBodyFactory.java delete mode 100644 java/common/src/org/apache/qpid/framing/AMQProtocolClassException.java delete mode 100644 java/common/src/org/apache/qpid/framing/AMQProtocolHeaderException.java delete mode 100644 java/common/src/org/apache/qpid/framing/AMQProtocolInstanceException.java delete mode 100644 java/common/src/org/apache/qpid/framing/AMQProtocolVersionException.java delete mode 100644 java/common/src/org/apache/qpid/framing/BasicContentHeaderProperties.java delete mode 100644 java/common/src/org/apache/qpid/framing/BodyFactory.java delete mode 100644 java/common/src/org/apache/qpid/framing/CompositeAMQDataBlock.java delete mode 100644 java/common/src/org/apache/qpid/framing/ContentBody.java delete mode 100644 java/common/src/org/apache/qpid/framing/ContentBodyFactory.java delete mode 100644 java/common/src/org/apache/qpid/framing/ContentHeaderBody.java delete mode 100644 java/common/src/org/apache/qpid/framing/ContentHeaderBodyFactory.java delete mode 100644 java/common/src/org/apache/qpid/framing/ContentHeaderProperties.java delete mode 100644 java/common/src/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java delete mode 100644 java/common/src/org/apache/qpid/framing/EncodableAMQDataBlock.java delete mode 100644 java/common/src/org/apache/qpid/framing/EncodingUtils.java delete mode 100644 java/common/src/org/apache/qpid/framing/FieldTable.java delete mode 100644 java/common/src/org/apache/qpid/framing/FieldTableKeyEnumeration.java delete mode 100644 java/common/src/org/apache/qpid/framing/HeartbeatBody.java delete mode 100644 java/common/src/org/apache/qpid/framing/HeartbeatBodyFactory.java delete mode 100644 java/common/src/org/apache/qpid/framing/ProtocolInitiation.java delete mode 100644 java/common/src/org/apache/qpid/pool/Event.java delete mode 100644 java/common/src/org/apache/qpid/pool/Job.java delete mode 100644 java/common/src/org/apache/qpid/pool/PoolingFilter.java delete mode 100644 java/common/src/org/apache/qpid/pool/ReadWriteThreadModel.java delete mode 100644 java/common/src/org/apache/qpid/pool/ReferenceCountingExecutorService.java delete mode 100644 java/common/src/org/apache/qpid/protocol/AMQConstant.java delete mode 100644 java/common/src/org/apache/qpid/ssl/BogusSSLContextFactory.java delete mode 100644 java/common/src/org/apache/qpid/ssl/BogusTrustManagerFactory.java delete mode 100644 java/common/src/org/apache/qpid/ssl/SSLServerSocketFactory.java delete mode 100644 java/common/src/org/apache/qpid/ssl/SSLSocketFactory.java delete mode 100644 java/common/src/org/apache/qpid/url/AMQBindingURL.java delete mode 100644 java/common/src/org/apache/qpid/url/BindingURL.java delete mode 100644 java/common/src/org/apache/qpid/url/URLHelper.java delete mode 100644 java/common/src/org/apache/qpid/url/URLSyntaxException.java delete mode 100644 java/common/src/org/apache/qpid/util/ConcurrentLinkedQueueAtomicSize.java delete mode 100644 java/common/src/org/apache/qpid/util/ConcurrentLinkedQueueNoSize.java (limited to 'java/common/src') diff --git a/java/common/src/main/java/org/apache/qpid/AMQChannelClosedException.java b/java/common/src/main/java/org/apache/qpid/AMQChannelClosedException.java new file mode 100644 index 0000000000..cd8b40c6da --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/AMQChannelClosedException.java @@ -0,0 +1,34 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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; + +/** + * AMQ channel closed exception. + */ +public class AMQChannelClosedException extends AMQException +{ + public AMQChannelClosedException(int errorCode, String msg) + { + super(errorCode, msg); + } +} + + diff --git a/java/common/src/main/java/org/apache/qpid/AMQChannelException.java b/java/common/src/main/java/org/apache/qpid/AMQChannelException.java new file mode 100644 index 0000000000..4d604f8c0b --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/AMQChannelException.java @@ -0,0 +1,49 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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; + +import org.apache.qpid.framing.ChannelCloseBody; +import org.apache.qpid.framing.AMQFrame; + +public class AMQChannelException extends AMQException +{ + private final int _classId; + private final int _methodId; + + public AMQChannelException(int errorCode, String msg, int classId, int methodId, Throwable t) + { + super(errorCode, msg, t); + _classId = classId; + _methodId = methodId; + } + + public AMQChannelException(int errorCode, String msg, int classId, int methodId) + { + super(errorCode, msg); + _classId = classId; + _methodId = methodId; + } + + public AMQFrame getCloseFrame(int channel) + { + return ChannelCloseBody.createAMQFrame(channel, getErrorCode(), getMessage(), _classId, _methodId); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/AMQConnectionClosedException.java b/java/common/src/main/java/org/apache/qpid/AMQConnectionClosedException.java new file mode 100644 index 0000000000..6ec18bad20 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/AMQConnectionClosedException.java @@ -0,0 +1,34 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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; + +/** + * AMQ channel closed exception. + */ +public class AMQConnectionClosedException extends AMQException +{ + public AMQConnectionClosedException(int errorCode, String msg) + { + super(errorCode, msg); + } +} + + diff --git a/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java b/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java new file mode 100644 index 0000000000..6254d80f32 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java @@ -0,0 +1,30 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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; + +public class AMQConnectionException extends AMQException +{ + public AMQConnectionException(String message) + { + super(message); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/AMQDisconnectedException.java b/java/common/src/main/java/org/apache/qpid/AMQDisconnectedException.java new file mode 100644 index 0000000000..b142eea73c --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/AMQDisconnectedException.java @@ -0,0 +1,34 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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; + +/** + * AMQ disconnected exception. + */ +public class AMQDisconnectedException extends AMQException +{ + public AMQDisconnectedException(String msg) + { + super(msg); + } +} + + diff --git a/java/common/src/main/java/org/apache/qpid/AMQException.java b/java/common/src/main/java/org/apache/qpid/AMQException.java new file mode 100644 index 0000000000..93c31e4fa8 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/AMQException.java @@ -0,0 +1,77 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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; + +import org.apache.log4j.Logger; + +/** + * Generic AMQ exception. + */ +public class AMQException extends Exception +{ + private int _errorCode; + + public AMQException(String message) + { + super(message); + } + + public AMQException(String msg, Throwable t) + { + super(msg, t); + } + + public AMQException(int errorCode, String msg, Throwable t) + { + super(msg + " [error code " + errorCode + ']', t); + _errorCode = errorCode; + } + + public AMQException(int errorCode, String msg) + { + super(msg + " [error code " + errorCode + ']'); + _errorCode = errorCode; + } + + public AMQException(Logger logger, String msg, Throwable t) + { + this(msg, t); + logger.error(getMessage(), this); + } + + public AMQException(Logger logger, String msg) + { + this(msg); + logger.error(getMessage(), this); + } + + public AMQException(Logger logger, int errorCode, String msg) + { + this(errorCode, msg); + logger.error(getMessage(), this); + } + + public int getErrorCode() + { + return _errorCode; + } + +} diff --git a/java/common/src/main/java/org/apache/qpid/AMQUndeliveredException.java b/java/common/src/main/java/org/apache/qpid/AMQUndeliveredException.java new file mode 100644 index 0000000000..4944ccc371 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/AMQUndeliveredException.java @@ -0,0 +1,44 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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; + +/** + * Generic AMQ exception. + */ +public class AMQUndeliveredException extends AMQException +{ + private Object _bounced; + + public AMQUndeliveredException(int errorCode, String msg, Object bounced) + { + super(errorCode, msg); + + _bounced = bounced; + } + + public Object getUndeliveredMessage() + { + return _bounced; + } + +} + + diff --git a/java/common/src/main/java/org/apache/qpid/AMQUnresolvedAddressException.java b/java/common/src/main/java/org/apache/qpid/AMQUnresolvedAddressException.java new file mode 100644 index 0000000000..2201903ded --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/AMQUnresolvedAddressException.java @@ -0,0 +1,37 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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; + +public class AMQUnresolvedAddressException extends AMQException +{ + String _broker; + + public AMQUnresolvedAddressException(String message, String broker) + { + super(message); + _broker = broker; + } + + public String toString() + { + return super.toString() + " Broker, \"" + _broker +"\""; + } +} diff --git a/java/common/src/main/java/org/apache/qpid/codec/AMQCodecFactory.java b/java/common/src/main/java/org/apache/qpid/codec/AMQCodecFactory.java new file mode 100644 index 0000000000..d7f1edbc30 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/codec/AMQCodecFactory.java @@ -0,0 +1,53 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.codec; + +import org.apache.mina.filter.codec.ProtocolCodecFactory; +import org.apache.mina.filter.codec.ProtocolDecoder; +import org.apache.mina.filter.codec.ProtocolEncoder; + +public class AMQCodecFactory implements ProtocolCodecFactory +{ + private AMQEncoder _encoder = new AMQEncoder(); + + private AMQDecoder _frameDecoder; + + /** + * @param expectProtocolInitiation true if the first frame received is going to be + * a protocol initiation frame, false if it is going to be a standard AMQ data block. + * The former case is used for the broker, which always expects to received the + * protocol initiation first from a newly connected client. + */ + public AMQCodecFactory(boolean expectProtocolInitiation) + { + _frameDecoder = new AMQDecoder(expectProtocolInitiation); + } + + public ProtocolEncoder getEncoder() + { + return _encoder; + } + + public ProtocolDecoder getDecoder() + { + return _frameDecoder; + } +} diff --git a/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java b/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java new file mode 100644 index 0000000000..bb981a242f --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java @@ -0,0 +1,99 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.codec; + +import org.apache.mina.common.ByteBuffer; +import org.apache.mina.common.IoSession; +import org.apache.mina.filter.codec.CumulativeProtocolDecoder; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.qpid.framing.AMQDataBlockDecoder; +import org.apache.qpid.framing.ProtocolInitiation; + +/** + * There is one instance of this class per session. Any changes or configuration done + * at run time to the encoders or decoders only affects decoding/encoding of the + * protocol session data to which is it bound. + * + */ +public class AMQDecoder extends CumulativeProtocolDecoder +{ + private AMQDataBlockDecoder _dataBlockDecoder = new AMQDataBlockDecoder(); + + private ProtocolInitiation.Decoder _piDecoder = new ProtocolInitiation.Decoder(); + + private boolean _expectProtocolInitiation; + + public AMQDecoder(boolean expectProtocolInitiation) + { + _expectProtocolInitiation = expectProtocolInitiation; + } + + protected boolean doDecode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception + { + if (_expectProtocolInitiation) + { + return doDecodePI(session, in, out); + } + else + { + return doDecodeDataBlock(session, in, out); + } + } + + protected boolean doDecodeDataBlock(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception + { + int pos = in.position(); + boolean enoughData = _dataBlockDecoder.decodable(session, in); + in.position(pos); + if (!enoughData) + { + // returning false means it will leave the contents in the buffer and + // call us again when more data has been read + return false; + } + else + { + _dataBlockDecoder.decode(session, in, out); + return true; + } + } + + private boolean doDecodePI(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception + { + boolean enoughData = _piDecoder.decodable(session, in); + if (!enoughData) + { + // returning false means it will leave the contents in the buffer and + // call us again when more data has been read + return false; + } + else + { + _piDecoder.decode(session, in, out); + return true; + } + } + + public void setExpectProtocolInitiation(boolean expectProtocolInitiation) + { + _expectProtocolInitiation = expectProtocolInitiation; + } +} diff --git a/java/common/src/main/java/org/apache/qpid/codec/AMQEncoder.java b/java/common/src/main/java/org/apache/qpid/codec/AMQEncoder.java new file mode 100644 index 0000000000..ad252aec35 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/codec/AMQEncoder.java @@ -0,0 +1,41 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.codec; + +import org.apache.mina.filter.codec.ProtocolEncoder; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.common.IoSession; +import org.apache.qpid.framing.AMQDataBlockEncoder; + +public class AMQEncoder implements ProtocolEncoder +{ + private AMQDataBlockEncoder _dataBlockEncoder = new AMQDataBlockEncoder(); + + public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws Exception + { + _dataBlockEncoder.encode(session, message, out); + } + + public void dispose(IoSession session) throws Exception + { + + } +} diff --git a/java/common/src/main/java/org/apache/qpid/configuration/Configured.java b/java/common/src/main/java/org/apache/qpid/configuration/Configured.java new file mode 100644 index 0000000000..2160dd6295 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/configuration/Configured.java @@ -0,0 +1,44 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.configuration; + +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.ElementType; +import java.lang.annotation.Target; + +/** + * Marks a field as being "configured" externally. + */ +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.FIELD) +public @interface Configured +{ + /** + * The Commons Configuration path to the configuration element + */ + String path(); + + /** + * The default value to use should the path not be found in the configuration source + */ + String defaultValue(); +} diff --git a/java/common/src/main/java/org/apache/qpid/configuration/PropertyException.java b/java/common/src/main/java/org/apache/qpid/configuration/PropertyException.java new file mode 100644 index 0000000000..12eebbf2b0 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/configuration/PropertyException.java @@ -0,0 +1,65 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.configuration; + +import org.apache.qpid.AMQException; +import org.apache.log4j.Logger; + +/** + * Indicates an error parsing a property expansion. + */ +public class PropertyException extends AMQException +{ + public PropertyException(String message) + { + super(message); + } + + public PropertyException(String msg, Throwable t) + { + super(msg, t); + } + + public PropertyException(int errorCode, String msg, Throwable t) + { + super(errorCode, msg, t); + } + + public PropertyException(int errorCode, String msg) + { + super(errorCode, msg); + } + + public PropertyException(Logger logger, String msg, Throwable t) + { + super(logger, msg, t); + } + + public PropertyException(Logger logger, String msg) + { + super(logger, msg); + } + + public PropertyException(Logger logger, int errorCode, String msg) + { + super(logger, errorCode, msg); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/configuration/PropertyUtils.java b/java/common/src/main/java/org/apache/qpid/configuration/PropertyUtils.java new file mode 100644 index 0000000000..37d8af2501 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/configuration/PropertyUtils.java @@ -0,0 +1,156 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.configuration; + +import java.util.ArrayList; +import java.util.Iterator; + +/** + * Based on code in Apache Ant, this utility class handles property expansion. This + * is most useful in config files and so on. + */ +public class PropertyUtils +{ + /** + * Replaces ${xxx} style constructions in the given value + * with the string value of the corresponding data types. Replaces only system + * properties + * + * @param value The string to be scanned for property references. + * May be null, in which case this + * method returns immediately with no effect. + * @return the original string with the properties replaced, or + * null if the original string is null. + * @throws PropertyException if the string contains an opening + * ${ without a closing + * } + */ + public static String replaceProperties(String value) throws PropertyException + { + if (value == null) + { + return null; + } + + ArrayList fragments = new ArrayList(); + ArrayList propertyRefs = new ArrayList(); + parsePropertyString(value, fragments, propertyRefs); + + StringBuffer sb = new StringBuffer(); + Iterator j = propertyRefs.iterator(); + + for (String fragment : fragments) + { + if (fragment == null) + { + String propertyName = (String) j.next(); + + // try to get it from the project or keys + // Backward compatibility + String replacement = System.getProperty(propertyName); + + if (replacement == null) + { + throw new PropertyException("Property ${" + propertyName + + "} has not been set"); + } + fragment = replacement; + } + sb.append(fragment); + } + + return sb.toString(); + } + + /** + * Default parsing method. Parses the supplied value for properties which are specified + * using ${foo} syntax. $X is left as is, and $$ specifies a single $. + * @param value the property string to parse + * @param fragments is populated with the string fragments. A null means "insert a + * property value here. The number of nulls in the list when populated is equal to the + * size of the propertyRefs list + * @param propertyRefs populated with the property names to be added into the final + * String. + */ + private static void parsePropertyString(String value, ArrayList fragments, + ArrayList propertyRefs) + throws PropertyException + { + int prev = 0; + int pos; + //search for the next instance of $ from the 'prev' position + while ((pos = value.indexOf("$", prev)) >= 0) + { + + //if there was any text before this, add it as a fragment + if (pos > 0) + { + fragments.add(value.substring(prev, pos)); + } + //if we are at the end of the string, we tack on a $ + //then move past it + if (pos == (value.length() - 1)) + { + fragments.add("$"); + prev = pos + 1; + } + else if (value.charAt(pos + 1) != '{') + { + //peek ahead to see if the next char is a property or not + //not a property: insert the char as a literal + if (value.charAt(pos + 1) == '$') + { + // two $ map to one $ + fragments.add("$"); + prev = pos + 2; + } + else + { + // $X maps to $X for all values of X!='$' + fragments.add(value.substring(pos, pos + 2)); + prev = pos + 2; + } + } + else + { + // property found, extract its name or bail on a typo + int endName = value.indexOf('}', pos); + if (endName < 0) + { + throw new PropertyException("Syntax error in property: " + + value); + } + String propertyName = value.substring(pos + 2, endName); + fragments.add(null); + propertyRefs.add(propertyName); + prev = endName + 1; + } + } + //no more $ signs found + //if there is any tail to the file, append it + if (prev < value.length()) + { + fragments.add(value.substring(prev)); + } + } + + +} diff --git a/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java b/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java new file mode 100644 index 0000000000..e67a5ba7fe --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java @@ -0,0 +1,36 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.exchange; + +public class ExchangeDefaults +{ + public final static String TOPIC_EXCHANGE_NAME = "amq.topic"; + + public final static String TOPIC_EXCHANGE_CLASS = "topic"; + + public final static String DIRECT_EXCHANGE_NAME = "amq.direct"; + + public final static String DIRECT_EXCHANGE_CLASS = "direct"; + + public final static String HEADERS_EXCHANGE_NAME = "amq.match"; + + public final static String HEADERS_EXCHANGE_CLASS = "headers"; +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQBody.java b/java/common/src/main/java/org/apache/qpid/framing/AMQBody.java new file mode 100644 index 0000000000..37efe1a34f --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQBody.java @@ -0,0 +1,38 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.mina.common.ByteBuffer; + +public abstract class AMQBody +{ + protected abstract byte getType(); + + /** + * Get the size of the body + * @return unsigned short + */ + protected abstract int getSize(); + + protected abstract void writePayload(ByteBuffer buffer); + + protected abstract void populateFromBuffer(ByteBuffer buffer, long size) throws AMQFrameDecodingException; +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlock.java b/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlock.java new file mode 100644 index 0000000000..9155c4024f --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlock.java @@ -0,0 +1,43 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.mina.common.ByteBuffer; + +/** + * A data block represents something that has a size in bytes and the ability to write itself to a byte + * buffer (similar to a byte array). + */ +public abstract class AMQDataBlock implements EncodableAMQDataBlock +{ + /** + * Get the size of buffer needed to store the byte representation of this + * frame. + * @return unsigned integer + */ + public abstract long getSize(); + + /** + * Writes the datablock to the specified buffer. + * @param buffer + */ + public abstract void writePayload(ByteBuffer buffer); +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockDecoder.java b/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockDecoder.java new file mode 100644 index 0000000000..438bfa8d82 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockDecoder.java @@ -0,0 +1,116 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.log4j.Logger; +import org.apache.mina.common.ByteBuffer; +import org.apache.mina.common.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; + +import java.util.HashMap; +import java.util.Map; + +public class AMQDataBlockDecoder +{ + Logger _logger = Logger.getLogger(AMQDataBlockDecoder.class); + + private final Map _supportedBodies = new HashMap(); + + public AMQDataBlockDecoder() + { + _supportedBodies.put(new Byte(AMQMethodBody.TYPE), AMQMethodBodyFactory.getInstance()); + _supportedBodies.put(new Byte(ContentHeaderBody.TYPE), ContentHeaderBodyFactory.getInstance()); + _supportedBodies.put(new Byte(ContentBody.TYPE), ContentBodyFactory.getInstance()); + _supportedBodies.put(new Byte(HeartbeatBody.TYPE), new HeartbeatBodyFactory()); + } + + public boolean decodable(IoSession session, ByteBuffer in) throws AMQFrameDecodingException + { + // type, channel, body size and end byte + if (in.remaining() < (1 + 2 + 4 + 1)) + { + return false; + } + + final byte type = in.get(); + final int channel = in.getUnsignedShort(); + final long bodySize = in.getUnsignedInt(); + + // bodySize can be zero + if (type <= 0 || channel < 0 || bodySize < 0) + { + throw new AMQFrameDecodingException("Undecodable frame: type = " + type + " channel = " + channel + + " bodySize = " + bodySize); + } + + if (in.remaining() < (bodySize + 1)) + { + return false; + } + return true; + } + + private boolean isSupportedFrameType(byte frameType) + { + final boolean result = _supportedBodies.containsKey(new Byte(frameType)); + + if (!result) + { + _logger.warn("AMQDataBlockDecoder does not handle frame type " + frameType); + } + + return result; + } + + protected Object createAndPopulateFrame(ByteBuffer in) + throws AMQFrameDecodingException + { + final byte type = in.get(); + if (!isSupportedFrameType(type)) + { + throw new AMQFrameDecodingException("Unsupported frame type: " + type); + } + final int channel = in.getUnsignedShort(); + final long bodySize = in.getUnsignedInt(); + + BodyFactory bodyFactory = (BodyFactory) _supportedBodies.get(new Byte(type)); + if (bodyFactory == null) + { + throw new AMQFrameDecodingException("Unsupported body type: " + type); + } + AMQFrame frame = new AMQFrame(); + + frame.populateFromBuffer(in, channel, bodySize, bodyFactory); + + byte marker = in.get(); + if ((marker & 0xFF) != 0xCE) + { + throw new AMQFrameDecodingException("End of frame marker not found. Read " + marker + " size=" + bodySize + " type=" + type); + } + return frame; + } + + public void decode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) + throws Exception + { + out.write(createAndPopulateFrame(in)); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockEncoder.java b/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockEncoder.java new file mode 100644 index 0000000000..3446563d35 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQDataBlockEncoder.java @@ -0,0 +1,65 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.log4j.Logger; +import org.apache.mina.common.ByteBuffer; +import org.apache.mina.common.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; + +import java.util.HashSet; +import java.util.Set; + +public class AMQDataBlockEncoder implements MessageEncoder +{ + Logger _logger = Logger.getLogger(AMQDataBlockEncoder.class); + + private Set _messageTypes; + + public AMQDataBlockEncoder() + { + _messageTypes = new HashSet(); + _messageTypes.add(EncodableAMQDataBlock.class); + } + + public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws Exception + { + final AMQDataBlock frame = (AMQDataBlock) message; + int frameSize = (int)frame.getSize(); + final ByteBuffer buffer = ByteBuffer.allocate(frameSize); + //buffer.setAutoExpand(true); + frame.writePayload(buffer); + + if (_logger.isDebugEnabled()) + { + _logger.debug("Encoded frame byte-buffer is '" + EncodingUtils.convertToHexString(buffer) + "'"); + } + + buffer.flip(); + out.write(buffer); + } + + public Set getMessageTypes() + { + return _messageTypes; + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQFrame.java b/java/common/src/main/java/org/apache/qpid/framing/AMQFrame.java new file mode 100644 index 0000000000..21c4bc8d5b --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQFrame.java @@ -0,0 +1,76 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.mina.common.ByteBuffer; + +public class AMQFrame extends AMQDataBlock implements EncodableAMQDataBlock +{ + public int channel; + + public AMQBody bodyFrame; + + public AMQFrame() + { + } + + public AMQFrame(int channel, AMQBody bodyFrame) + { + this.channel = channel; + this.bodyFrame = bodyFrame; + } + + public long getSize() + { + return 1 + 2 + 4 + bodyFrame.getSize() + 1; + } + + public void writePayload(ByteBuffer buffer) + { + buffer.put(bodyFrame.getType()); + // TODO: how does channel get populated + EncodingUtils.writeUnsignedShort(buffer, channel); + EncodingUtils.writeUnsignedInteger(buffer, bodyFrame.getSize()); + bodyFrame.writePayload(buffer); + buffer.put((byte) 0xCE); + } + + /** + * + * @param buffer + * @param channel unsigned short + * @param bodySize unsigned integer + * @param bodyFactory + * @throws AMQFrameDecodingException + */ + public void populateFromBuffer(ByteBuffer buffer, int channel, long bodySize, BodyFactory bodyFactory) + throws AMQFrameDecodingException + { + this.channel = channel; + bodyFrame = bodyFactory.createBody(buffer); + bodyFrame.populateFromBuffer(buffer, bodySize); + } + + public String toString() + { + return "Frame channelId: " + channel + ", bodyFrame: " + String.valueOf(bodyFrame); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQFrameDecodingException.java b/java/common/src/main/java/org/apache/qpid/framing/AMQFrameDecodingException.java new file mode 100644 index 0000000000..a24bd6aaa9 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQFrameDecodingException.java @@ -0,0 +1,48 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.log4j.Logger; +import org.apache.qpid.AMQException; + +public class AMQFrameDecodingException extends AMQException +{ + public AMQFrameDecodingException(String message) + { + super(message); + } + + public AMQFrameDecodingException(String message, Throwable t) + { + super(message, t); + } + + public AMQFrameDecodingException(Logger log, String message) + { + super(log, message); + } + + public AMQFrameDecodingException(Logger log, String message, Throwable t) + { + super(log, message, t); + } + +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBody.java b/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBody.java new file mode 100644 index 0000000000..c3e02aed56 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBody.java @@ -0,0 +1,90 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.mina.common.ByteBuffer; +import org.apache.qpid.AMQChannelException; + +public abstract class AMQMethodBody extends AMQBody +{ + public static final byte TYPE = 1; + + /** unsigned short */ + protected abstract int getBodySize(); + + /** + * @return unsigned short + */ + protected abstract int getClazz(); + + /** + * @return unsigned short + */ + protected abstract int getMethod(); + + protected abstract void writeMethodPayload(ByteBuffer buffer); + + protected byte getType() + { + return TYPE; + } + + protected int getSize() + { + return 2 + 2 + getBodySize(); + } + + protected void writePayload(ByteBuffer buffer) + { + EncodingUtils.writeUnsignedShort(buffer, getClazz()); + EncodingUtils.writeUnsignedShort(buffer, getMethod()); + writeMethodPayload(buffer); + } + + protected abstract void populateMethodBodyFromBuffer(ByteBuffer buffer) throws AMQFrameDecodingException; + + protected void populateFromBuffer(ByteBuffer buffer, long size) throws AMQFrameDecodingException + { + populateMethodBodyFromBuffer(buffer); + } + + public String toString() + { + StringBuffer buf = new StringBuffer(getClass().toString()); + buf.append(" Class: ").append(getClazz()); + buf.append(" Method: ").append(getMethod()); + return buf.toString(); + } + + /** + * Creates an AMQChannelException for the corresponding body type (a channel exception + * should include the class and method ids of the body it resulted from). + */ + public AMQChannelException getChannelException(int code, String message) + { + return new AMQChannelException(code, message, getClazz(), getMethod()); + } + + public AMQChannelException getChannelException(int code, String message, Throwable cause) + { + return new AMQChannelException(code, message, getClazz(), getMethod(), cause); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBodyFactory.java b/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBodyFactory.java new file mode 100644 index 0000000000..107af67dc7 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBodyFactory.java @@ -0,0 +1,46 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.log4j.Logger; +import org.apache.mina.common.ByteBuffer; + +public class AMQMethodBodyFactory implements BodyFactory +{ + private static final Logger _log = Logger.getLogger(AMQMethodBodyFactory.class); + + private static final AMQMethodBodyFactory _instance = new AMQMethodBodyFactory(); + + public static AMQMethodBodyFactory getInstance() + { + return _instance; + } + + private AMQMethodBodyFactory() + { + _log.debug("Creating method body factory"); + } + + public AMQBody createBody(ByteBuffer in) throws AMQFrameDecodingException + { + return MethodBodyDecoderRegistry.get(in.getUnsignedShort(), in.getUnsignedShort()); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolClassException.java b/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolClassException.java new file mode 100644 index 0000000000..e6acad502f --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolClassException.java @@ -0,0 +1,29 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +public class AMQProtocolClassException extends AMQProtocolHeaderException +{ + public AMQProtocolClassException(String message) + { + super(message); + } +} \ No newline at end of file diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolHeaderException.java b/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolHeaderException.java new file mode 100644 index 0000000000..888ed14faf --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolHeaderException.java @@ -0,0 +1,31 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.qpid.AMQException; + +public class AMQProtocolHeaderException extends AMQException +{ + public AMQProtocolHeaderException(String message) + { + super(message); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolInstanceException.java b/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolInstanceException.java new file mode 100644 index 0000000000..c58979f876 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolInstanceException.java @@ -0,0 +1,29 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +public class AMQProtocolInstanceException extends AMQProtocolHeaderException +{ + public AMQProtocolInstanceException(String message) + { + super(message); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolVersionException.java b/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolVersionException.java new file mode 100644 index 0000000000..7b326a0dc4 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolVersionException.java @@ -0,0 +1,33 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +/** + * Exception that is thrown when the client and server differ on expected protocol version (header) information. + * + */ +public class AMQProtocolVersionException extends AMQProtocolHeaderException +{ + public AMQProtocolVersionException(String message) + { + super(message); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java b/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java new file mode 100644 index 0000000000..3c7e656053 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java @@ -0,0 +1,592 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.log4j.Logger; +import org.apache.mina.common.ByteBuffer; + +public class BasicContentHeaderProperties implements ContentHeaderProperties +{ + private static final Logger _logger = Logger.getLogger(BasicContentHeaderProperties.class); + + /** + * We store the encoded form when we decode the content header so that if we need to + * write it out without modifying it we can do so without incurring the expense of + * reencoding it + */ + private byte[] _encodedForm; + + /** + * Flag indicating whether the entire content header has been decoded yet + */ + private boolean _decoded = true; + + /** + * We have some optimisations for partial decoding for maximum performance. The headers are used in the broker + * for routing in some cases so we can decode that separately. + */ + private boolean _decodedHeaders = true; + + /** + * We have some optimisations for partial decoding for maximum performance. The content type is used by all + * clients to determine the message type + */ + private boolean _decodedContentType = true; + + private String _contentType; + + private String _encoding; + + private FieldTable _headers; + + private byte _deliveryMode; + + private byte _priority; + + private String _correlationId; + + private String _replyTo; + + private long _expiration; + + private String _messageId; + + private long _timestamp; + + private String _type; + + private String _userId; + + private String _appId; + + private String _clusterId; + + private int _propertyFlags = 0; + + public BasicContentHeaderProperties() + { + } + + public int getPropertyListSize() + { + if (_encodedForm != null) + { + return _encodedForm.length; + } + else + { + int size = 0; + + if ((_propertyFlags & (1 << 15)) > 0) + { + size += EncodingUtils.encodedShortStringLength(_contentType); + } + if ((_propertyFlags & (1 << 14)) > 0) + { + size += EncodingUtils.encodedShortStringLength(_encoding); + } + if ((_propertyFlags & (1 << 13)) > 0) + { + size += EncodingUtils.encodedFieldTableLength(_headers); + } + if ((_propertyFlags & (1 << 12)) > 0) + { + size += 1; + } + if ((_propertyFlags & (1 << 11)) > 0) + { + size += 1; + } + if ((_propertyFlags & (1 << 10)) > 0) + { + size += EncodingUtils.encodedShortStringLength(_correlationId); + } + if ((_propertyFlags & (1 << 9)) > 0) + { + size += EncodingUtils.encodedShortStringLength(_replyTo); + } + if ((_propertyFlags & (1 << 8)) > 0) + { + size += EncodingUtils.encodedShortStringLength(String.valueOf(_expiration)); + } + if ((_propertyFlags & (1 << 7)) > 0) + { + size += EncodingUtils.encodedShortStringLength(_messageId); + } + if ((_propertyFlags & (1 << 6)) > 0) + { + size += 8; + } + if ((_propertyFlags & (1 << 5)) > 0) + { + size += EncodingUtils.encodedShortStringLength(_type); + } + if ((_propertyFlags & (1 << 4)) > 0) + { + size += EncodingUtils.encodedShortStringLength(_userId); + } + if ((_propertyFlags & (1 << 3)) > 0) + { + size += EncodingUtils.encodedShortStringLength(_appId); + } + if ((_propertyFlags & (1 << 2)) > 0) + { + size += EncodingUtils.encodedShortStringLength(_clusterId); + } + return size; + } + } + + private void clearEncodedForm() + { + if (!_decoded && _encodedForm != null) + { + //decode(); + } + _encodedForm = null; + } + + public void setPropertyFlags(int propertyFlags) + { + clearEncodedForm(); + _propertyFlags = propertyFlags; + } + + public int getPropertyFlags() + { + return _propertyFlags; + } + + public void writePropertyListPayload(ByteBuffer buffer) + { + if (_encodedForm != null) + { + buffer.put(_encodedForm); + } + else + { + if ((_propertyFlags & (1 << 15)) > 0) + { + EncodingUtils.writeShortStringBytes(buffer, _contentType); + } + if ((_propertyFlags & (1 << 14)) > 0) + { + EncodingUtils.writeShortStringBytes(buffer, _encoding); + } + if ((_propertyFlags & (1 << 13)) > 0) + { + EncodingUtils.writeFieldTableBytes(buffer, _headers); + } + if ((_propertyFlags & (1 << 12)) > 0) + { + buffer.put(_deliveryMode); + } + if ((_propertyFlags & (1 << 11)) > 0) + { + buffer.put(_priority); + } + if ((_propertyFlags & (1 << 10)) > 0) + { + EncodingUtils.writeShortStringBytes(buffer, _correlationId); + } + if ((_propertyFlags & (1 << 9)) > 0) + { + EncodingUtils.writeShortStringBytes(buffer, _replyTo); + } + if ((_propertyFlags & (1 << 8)) > 0) + { + EncodingUtils.writeShortStringBytes(buffer, String.valueOf(_expiration)); + } + if ((_propertyFlags & (1 << 7)) > 0) + { + EncodingUtils.writeShortStringBytes(buffer, _messageId); + } + if ((_propertyFlags & (1 << 6)) > 0) + { + EncodingUtils.writeTimestamp(buffer, _timestamp); + } + if ((_propertyFlags & (1 << 5)) > 0) + { + EncodingUtils.writeShortStringBytes(buffer, _type); + } + if ((_propertyFlags & (1 << 4)) > 0) + { + EncodingUtils.writeShortStringBytes(buffer, _userId); + } + if ((_propertyFlags & (1 << 3)) > 0) + { + EncodingUtils.writeShortStringBytes(buffer, _appId); + } + if ((_propertyFlags & (1 << 2)) > 0) + { + EncodingUtils.writeShortStringBytes(buffer, _clusterId); + } + } + } + + public void populatePropertiesFromBuffer(ByteBuffer buffer, int propertyFlags, int size) + throws AMQFrameDecodingException + { + _propertyFlags = propertyFlags; + + if (_logger.isDebugEnabled()) + { + _logger.debug("Property flags: " + _propertyFlags); + } + decode(buffer); + /*_encodedForm = new byte[size]; + buffer.get(_encodedForm, 0, size); + _decoded = false; + _decodedHeaders = false; + _decodedContentType = false;*/ + } + + private void decode(ByteBuffer buffer) + { + //ByteBuffer buffer = ByteBuffer.wrap(_encodedForm); + int pos = buffer.position(); + try + { + if ((_propertyFlags & (1 << 15)) > 0) + { + _contentType = EncodingUtils.readShortString(buffer); + } + if ((_propertyFlags & (1 << 14)) > 0) + { + _encoding = EncodingUtils.readShortString(buffer); + } + if ((_propertyFlags & (1 << 13)) > 0) + { + _headers = EncodingUtils.readFieldTable(buffer); + } + if ((_propertyFlags & (1 << 12)) > 0) + { + _deliveryMode = buffer.get(); + } + if ((_propertyFlags & (1 << 11)) > 0) + { + _priority = buffer.get(); + } + if ((_propertyFlags & (1 << 10)) > 0) + { + _correlationId = EncodingUtils.readShortString(buffer); + } + if ((_propertyFlags & (1 << 9)) > 0) + { + _replyTo = EncodingUtils.readShortString(buffer); + } + if ((_propertyFlags & (1 << 8)) > 0) + { + _expiration = Long.parseLong(EncodingUtils.readShortString(buffer)); + } + if ((_propertyFlags & (1 << 7)) > 0) + { + _messageId = EncodingUtils.readShortString(buffer); + } + if ((_propertyFlags & (1 << 6)) > 0) + { + _timestamp = EncodingUtils.readTimestamp(buffer); + } + if ((_propertyFlags & (1 << 5)) > 0) + { + _type = EncodingUtils.readShortString(buffer); + } + if ((_propertyFlags & (1 << 4)) > 0) + { + _userId = EncodingUtils.readShortString(buffer); + } + if ((_propertyFlags & (1 << 3)) > 0) + { + _appId = EncodingUtils.readShortString(buffer); + } + if ((_propertyFlags & (1 << 2)) > 0) + { + _clusterId = EncodingUtils.readShortString(buffer); + } + } + catch (AMQFrameDecodingException e) + { + throw new RuntimeException("Error in content header data: " + e); + } + + final int endPos = buffer.position(); + buffer.position(pos); + final int len = endPos - pos; + _encodedForm = new byte[len]; + final int limit = buffer.limit(); + buffer.limit(endPos); + buffer.get(_encodedForm, 0, len); + buffer.limit(limit); + buffer.position(endPos); + _decoded = true; + } + + + private void decodeUpToHeaders() + { + ByteBuffer buffer = ByteBuffer.wrap(_encodedForm); + try + { + if ((_propertyFlags & (1 << 15)) > 0) + { + byte length = buffer.get(); + buffer.skip(length); + } + if ((_propertyFlags & (1 << 14)) > 0) + { + byte length = buffer.get(); + buffer.skip(length); + } + if ((_propertyFlags & (1 << 13)) > 0) + { + _headers = EncodingUtils.readFieldTable(buffer); + } + _decodedHeaders = true; + } + catch (AMQFrameDecodingException e) + { + throw new RuntimeException("Error in content header data: " + e); + } + } + + private void decodeUpToContentType() + { + ByteBuffer buffer = ByteBuffer.wrap(_encodedForm); + + if ((_propertyFlags & (1 << 15)) > 0) + { + _contentType = EncodingUtils.readShortString(buffer); + } + + _decodedContentType = true; + } + + private void decodeIfNecessary() + { + if (!_decoded) + { + //decode(); + } + } + + private void decodeHeadersIfNecessary() + { + if (!_decoded && !_decodedHeaders) + { + decodeUpToHeaders(); + } + } + + private void decodeContentTypeIfNecessary() + { + if (!_decoded && !_decodedContentType) + { + decodeUpToContentType(); + } + } + public String getContentType() + { + decodeContentTypeIfNecessary(); + return _contentType; + } + + public void setContentType(String contentType) + { + clearEncodedForm(); + _propertyFlags |= (1 << 15); + _contentType = contentType; + } + + public String getEncoding() + { + decodeIfNecessary(); + return _encoding; + } + + public void setEncoding(String encoding) + { + clearEncodedForm(); + _propertyFlags |= (1 << 14); + _encoding = encoding; + } + + public FieldTable getHeaders() + { + decodeHeadersIfNecessary(); + return _headers; + } + + public void setHeaders(FieldTable headers) + { + clearEncodedForm(); + _propertyFlags |= (1 << 13); + _headers = headers; + } + + public byte getDeliveryMode() + { + decodeIfNecessary(); + return _deliveryMode; + } + + public void setDeliveryMode(byte deliveryMode) + { + clearEncodedForm(); + _propertyFlags |= (1 << 12); + _deliveryMode = deliveryMode; + } + + public byte getPriority() + { + decodeIfNecessary(); + return _priority; + } + + public void setPriority(byte priority) + { + clearEncodedForm(); + _propertyFlags |= (1 << 11); + _priority = priority; + } + + public String getCorrelationId() + { + decodeIfNecessary(); + return _correlationId; + } + + public void setCorrelationId(String correlationId) + { + clearEncodedForm(); + _propertyFlags |= (1 << 10); + _correlationId = correlationId; + } + + public String getReplyTo() + { + decodeIfNecessary(); + return _replyTo; + } + + public void setReplyTo(String replyTo) + { + clearEncodedForm(); + _propertyFlags |= (1 << 9); + _replyTo = replyTo; + } + + public long getExpiration() + { + decodeIfNecessary(); + return _expiration; + } + + public void setExpiration(long expiration) + { + clearEncodedForm(); + _propertyFlags |= (1 << 8); + _expiration = expiration; + } + + + public String getMessageId() + { + decodeIfNecessary(); + return _messageId; + } + + public void setMessageId(String messageId) + { + clearEncodedForm(); + _propertyFlags |= (1 << 7); + _messageId = messageId; + } + + public long getTimestamp() + { + decodeIfNecessary(); + return _timestamp; + } + + public void setTimestamp(long timestamp) + { + clearEncodedForm(); + _propertyFlags |= (1 << 6); + _timestamp = timestamp; + } + + public String getType() + { + decodeIfNecessary(); + return _type; + } + + public void setType(String type) + { + clearEncodedForm(); + _propertyFlags |= (1 << 5); + _type = type; + } + + public String getUserId() + { + decodeIfNecessary(); + return _userId; + } + + public void setUserId(String userId) + { + clearEncodedForm(); + _propertyFlags |= (1 << 4); + _userId = userId; + } + + public String getAppId() + { + decodeIfNecessary(); + return _appId; + } + + public void setAppId(String appId) + { + clearEncodedForm(); + _propertyFlags |= (1 << 3); + _appId = appId; + } + + public String getClusterId() + { + decodeIfNecessary(); + return _clusterId; + } + + public void setClusterId(String clusterId) + { + clearEncodedForm(); + _propertyFlags |= (1 << 2); + _clusterId = clusterId; + } + + public String toString() + { + return "reply-to = " + _replyTo + " propertyFlags = " + _propertyFlags; + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/BodyFactory.java b/java/common/src/main/java/org/apache/qpid/framing/BodyFactory.java new file mode 100644 index 0000000000..cf5708d993 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/BodyFactory.java @@ -0,0 +1,31 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.mina.common.ByteBuffer; + +/** + * Any class that is capable of turning a stream of bytes into an AMQ structure must implement this interface. + */ +public interface BodyFactory +{ + AMQBody createBody(ByteBuffer in) throws AMQFrameDecodingException; +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java b/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java new file mode 100644 index 0000000000..5ec62ede93 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java @@ -0,0 +1,103 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.mina.common.ByteBuffer; + +public class CompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQDataBlock +{ + private ByteBuffer _encodedBlock; + + private AMQDataBlock[] _blocks; + + public CompositeAMQDataBlock(AMQDataBlock[] blocks) + { + _blocks = blocks; + } + + /** + * The encoded block will be logically first before the AMQDataBlocks which are encoded + * into the buffer afterwards. + * @param encodedBlock already-encoded data + * @param blocks some blocks to be encoded. + */ + public CompositeAMQDataBlock(ByteBuffer encodedBlock, AMQDataBlock[] blocks) + { + this(blocks); + _encodedBlock = encodedBlock; + } + + public AMQDataBlock[] getBlocks() + { + return _blocks; + } + + public ByteBuffer getEncodedBlock() + { + return _encodedBlock; + } + + public long getSize() + { + long frameSize = 0; + for (int i = 0; i < _blocks.length; i++) + { + frameSize += _blocks[i].getSize(); + } + if (_encodedBlock != null) + { + _encodedBlock.rewind(); + frameSize += _encodedBlock.remaining(); + } + return frameSize; + } + + public void writePayload(ByteBuffer buffer) + { + if (_encodedBlock != null) + { + buffer.put(_encodedBlock); + } + for (int i = 0; i < _blocks.length; i++) + { + _blocks[i].writePayload(buffer); + } + } + + public String toString() + { + if (_blocks == null) + { + return "No blocks contained in composite frame"; + } + else + { + StringBuilder buf = new StringBuilder(this.getClass().getName()); + buf.append("{encodedBlock=").append(_encodedBlock); + for (int i = 0 ; i < _blocks.length; i++) + { + buf.append(" ").append(i).append("=[").append(_blocks[i].toString()).append("]"); + } + buf.append("}"); + return buf.toString(); + } + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/ContentBody.java b/java/common/src/main/java/org/apache/qpid/framing/ContentBody.java new file mode 100644 index 0000000000..2aae833fcd --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/ContentBody.java @@ -0,0 +1,85 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.mina.common.ByteBuffer; + +public class ContentBody extends AMQBody +{ + public static final byte TYPE = 3; + + public ByteBuffer payload; + + protected byte getType() + { + return TYPE; + } + + public int getSize() + { + return (payload == null ? 0 : payload.limit()); + } + + public void writePayload(ByteBuffer buffer) + { + if (payload != null) + { + ByteBuffer copy = payload.duplicate(); + buffer.put(copy.rewind()); + } + } + + protected void populateFromBuffer(ByteBuffer buffer, long size) throws AMQFrameDecodingException + { + if (size > 0) + { + payload = buffer.slice(); + payload.limit((int) size); + buffer.skip((int) size); + } + + } + + public void reduceBufferToFit() + { + if (payload != null && (payload.remaining() < payload.capacity() / 2)) + { + int size = payload.limit(); + ByteBuffer newPayload = ByteBuffer.allocate(size); + + newPayload.put(payload); + newPayload.flip(); + + //reduce reference count on payload + payload.release(); + + payload = newPayload; + } + } + + public static AMQFrame createAMQFrame(int channelId, ContentBody body) + { + final AMQFrame frame = new AMQFrame(); + frame.channel = channelId; + frame.bodyFrame = body; + return frame; + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/ContentBodyFactory.java b/java/common/src/main/java/org/apache/qpid/framing/ContentBodyFactory.java new file mode 100644 index 0000000000..22af331ab7 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/ContentBodyFactory.java @@ -0,0 +1,47 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.log4j.Logger; +import org.apache.mina.common.ByteBuffer; + +public class ContentBodyFactory implements BodyFactory +{ + private static final Logger _log = Logger.getLogger(AMQMethodBodyFactory.class); + + private static final ContentBodyFactory _instance = new ContentBodyFactory(); + + public static ContentBodyFactory getInstance() + { + return _instance; + } + + private ContentBodyFactory() + { + _log.debug("Creating content body factory"); + } + + public AMQBody createBody(ByteBuffer in) throws AMQFrameDecodingException + { + return new ContentBody(); + } +} + diff --git a/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java b/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java new file mode 100644 index 0000000000..46b933b2c3 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java @@ -0,0 +1,115 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.mina.common.ByteBuffer; + +public class ContentHeaderBody extends AMQBody +{ + public static final byte TYPE = 2; + + public int classId; + + public int weight; + + /** unsigned long but java can't handle that anyway when allocating byte array */ + public long bodySize; + + /** must never be null */ + public ContentHeaderProperties properties; + + public ContentHeaderBody() + { + } + + public ContentHeaderBody(ContentHeaderProperties props, int classId) + { + properties = props; + this.classId = classId; + } + + public ContentHeaderBody(int classId, int weight, ContentHeaderProperties props, long bodySize) + { + this(props, classId); + this.weight = weight; + this.bodySize = bodySize; + } + + protected byte getType() + { + return TYPE; + } + + protected void populateFromBuffer(ByteBuffer buffer, long size) throws AMQFrameDecodingException + { + classId = buffer.getUnsignedShort(); + weight = buffer.getUnsignedShort(); + bodySize = buffer.getLong(); + int propertyFlags = buffer.getUnsignedShort(); + ContentHeaderPropertiesFactory factory = ContentHeaderPropertiesFactory.getInstance(); + properties = factory.createContentHeaderProperties(classId, propertyFlags, buffer, (int)size - 14); + } + + /** + * Helper method that is used currently by the persistence layer (by BDB at the moment). + * @param buffer + * @param size + * @return + * @throws AMQFrameDecodingException + */ + public static ContentHeaderBody createFromBuffer(ByteBuffer buffer, long size) throws AMQFrameDecodingException + { + ContentHeaderBody body = new ContentHeaderBody(); + body.populateFromBuffer(buffer, size); + return body; + } + + public int getSize() + { + return 2 + 2 + 8 + 2 + properties.getPropertyListSize(); + } + + public void writePayload(ByteBuffer buffer) + { + EncodingUtils.writeUnsignedShort(buffer, classId); + EncodingUtils.writeUnsignedShort(buffer, weight); + buffer.putLong(bodySize); + EncodingUtils.writeUnsignedShort(buffer, properties.getPropertyFlags()); + properties.writePropertyListPayload(buffer); + } + + public static AMQFrame createAMQFrame(int channelId, int classId, int weight, BasicContentHeaderProperties properties, + long bodySize) + { + final AMQFrame frame = new AMQFrame(); + frame.channel = channelId; + frame.bodyFrame = new ContentHeaderBody(classId, weight, properties, bodySize); + return frame; + } + + public static AMQFrame createAMQFrame(int channelId, ContentHeaderBody body) + { + final AMQFrame frame = new AMQFrame(); + frame.channel = channelId; + frame.bodyFrame = body; + return frame; + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBodyFactory.java b/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBodyFactory.java new file mode 100644 index 0000000000..ddf63f8aa3 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBodyFactory.java @@ -0,0 +1,50 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.log4j.Logger; +import org.apache.mina.common.ByteBuffer; + +public class ContentHeaderBodyFactory implements BodyFactory +{ + private static final Logger _log = Logger.getLogger(AMQMethodBodyFactory.class); + + private static final ContentHeaderBodyFactory _instance = new ContentHeaderBodyFactory(); + + public static ContentHeaderBodyFactory getInstance() + { + return _instance; + } + + private ContentHeaderBodyFactory() + { + _log.debug("Creating content header body factory"); + } + + public AMQBody createBody(ByteBuffer in) throws AMQFrameDecodingException + { + // all content headers are the same - it is only the properties that differ. + // the content header body further delegates construction of properties + return new ContentHeaderBody(); + } + + +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderProperties.java b/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderProperties.java new file mode 100644 index 0000000000..561d7852fd --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderProperties.java @@ -0,0 +1,58 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.mina.common.ByteBuffer; + +/** + * There will be an implementation of this interface for each content type. All content types have associated + * header properties and this provides a way to encode and decode them. + */ +public interface ContentHeaderProperties +{ + /** + * Writes the property list to the buffer, in a suitably encoded form. + * @param buffer The buffer to write to + */ + void writePropertyListPayload(ByteBuffer buffer); + + /** + * Populates the properties from buffer. + * @param buffer The buffer to read from. + * @param propertyFlags he property flags. + * @throws AMQFrameDecodingException when the buffer does not contain valid data + */ + void populatePropertiesFromBuffer(ByteBuffer buffer, int propertyFlags, int size) + throws AMQFrameDecodingException; + + /** + * @return the size of the encoded property list in bytes. + */ + int getPropertyListSize(); + + /** + * Gets the property flags. Property flags indicate which properties are set in the list. The + * position and meaning of each flag is defined in the protocol specification for the particular + * content type with which these properties are associated. + * @return flags + */ + int getPropertyFlags(); +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java b/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java new file mode 100644 index 0000000000..cec413cb9d --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java @@ -0,0 +1,54 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.mina.common.ByteBuffer; + +public class ContentHeaderPropertiesFactory +{ + private static final ContentHeaderPropertiesFactory _instance = new ContentHeaderPropertiesFactory(); + + public static ContentHeaderPropertiesFactory getInstance() + { + return _instance; + } + + private ContentHeaderPropertiesFactory() + { + } + + public ContentHeaderProperties createContentHeaderProperties(int classId, int propertyFlags, + ByteBuffer buffer, int size) + throws AMQFrameDecodingException + { + ContentHeaderProperties properties; + switch (classId) + { + case BasicConsumeBody.CLASS_ID: + properties = new BasicContentHeaderProperties(); + break; + default: + throw new AMQFrameDecodingException("Unsupport content header class id: " + classId); + } + properties.populatePropertiesFromBuffer(buffer, propertyFlags, size); + return properties; + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/EncodableAMQDataBlock.java b/java/common/src/main/java/org/apache/qpid/framing/EncodableAMQDataBlock.java new file mode 100644 index 0000000000..9cf96e698c --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/EncodableAMQDataBlock.java @@ -0,0 +1,35 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +/** + * Marker interface to indicate to MINA that a data block should be encoded with the + * single encoder/decoder that we have defined. + * + * Note that due to a bug in MINA all classes must directly implement this interface, even if + * a superclass implements it. + * TODO: fix MINA so that this is not necessary + * + */ +public interface EncodableAMQDataBlock +{ + +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java b/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java new file mode 100644 index 0000000000..3a683b8e90 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java @@ -0,0 +1,549 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.log4j.Logger; +import org.apache.mina.common.ByteBuffer; + +import java.nio.charset.Charset; + +public class EncodingUtils +{ + private static final Logger _logger = Logger.getLogger(EncodingUtils.class); + + private static final String STRING_ENCODING = "iso8859-15"; + + private static final Charset _charset = Charset.forName("iso8859-15"); + + public static final int SIZEOF_UNSIGNED_SHORT = 2; + public static final int SIZEOF_UNSIGNED_INT = 4; + + public static int encodedShortStringLength(String s) + { + if (s == null) + { + return 1; + } + else + { + return (short) (1 + s.length()); + } + } + + public static int encodedLongStringLength(String s) + { + if (s == null) + { + return 4; + } + else + { + return 4 + s.length(); + } + } + + public static int encodedLongStringLength(char[] s) + { + if (s == null) + { + return 4; + } + else + { + return 4 + s.length; + } + } + + public static int encodedLongstrLength(byte[] bytes) + { + if (bytes == null) + { + return 4; + } + else + { + return 4 + bytes.length; + } + } + + public static int encodedFieldTableLength(FieldTable table) + { + if (table == null) + { + // size is encoded as 4 octets + return 4; + } + else + { + // size of the table plus 4 octets for the size + return (int) table.getEncodedSize() + 4; + } + } + + public static void writeShortStringBytes(ByteBuffer buffer, String s) + { + if (s != null) + { + byte[] encodedString = new byte[s.length()]; + char[] cha = s.toCharArray(); + for (int i = 0; i < cha.length; i++) + { + encodedString[i] = (byte) cha[i]; + } + // TODO: check length fits in an unsigned byte + writeUnsignedByte(buffer, (short) encodedString.length); + buffer.put(encodedString); + } + else + { + // really writing out unsigned byte + buffer.put((byte) 0); + } + } + + public static void writeLongStringBytes(ByteBuffer buffer, String s) + { + assert s == null || s.length() <= 0xFFFE; + if (s != null) + { + int len = s.length(); + writeUnsignedInteger(buffer, s.length()); + byte[] encodedString = new byte[len]; + char[] cha = s.toCharArray(); + for (int i = 0; i < cha.length; i++) + { + encodedString[i] = (byte) cha[i]; + } + buffer.put(encodedString); + } + else + { + writeUnsignedInteger(buffer, 0); + } + } + + public static void writeLongStringBytes(ByteBuffer buffer, char[] s) + { + assert s == null || s.length <= 0xFFFE; + if (s != null) + { + int len = s.length; + writeUnsignedInteger(buffer, s.length); + byte[] encodedString = new byte[len]; + for (int i = 0; i < s.length; i++) + { + encodedString[i] = (byte) s[i]; + } + buffer.put(encodedString); + } + else + { + writeUnsignedInteger(buffer, 0); + } + } + + public static void writeLongStringBytes(ByteBuffer buffer, byte[] bytes) + { + assert bytes == null || bytes.length <= 0xFFFE; + if (bytes != null) + { + writeUnsignedInteger(buffer, bytes.length); + buffer.put(bytes); + } + else + { + writeUnsignedInteger(buffer, 0); + } + } + + public static void writeUnsignedByte(ByteBuffer buffer, short b) + { + byte bv = (byte) b; + buffer.put(bv); + } + + public static void writeUnsignedShort(ByteBuffer buffer, int s) + { + // TODO: Is this comparison safe? Do I need to cast RHS to long? + if (s < Short.MAX_VALUE) + { + buffer.putShort((short) s); + } + else + { + short sv = (short) s; + buffer.put((byte) (0xFF & (sv >> 8))); + buffer.put((byte) (0xFF & sv)); + } + } + + public static void writeUnsignedInteger(ByteBuffer buffer, long l) + { + // TODO: Is this comparison safe? Do I need to cast RHS to long? + if (l < Integer.MAX_VALUE) + { + buffer.putInt((int) l); + } + else + { + int iv = (int) l; + + // FIXME: This *may* go faster if we build this into a local 4-byte array and then + // put the array in a single call. + buffer.put((byte) (0xFF & (iv >> 24))); + buffer.put((byte) (0xFF & (iv >> 16))); + buffer.put((byte) (0xFF & (iv >> 8))); + buffer.put((byte) (0xFF & iv)); + } + } + + public static void writeFieldTableBytes(ByteBuffer buffer, FieldTable table) + { + if (table != null) + { + table.writeToBuffer(buffer); + } + else + { + EncodingUtils.writeUnsignedInteger(buffer, 0); + } + } + + public static void writeBooleans(ByteBuffer buffer, boolean[] values) + { + byte packedValue = 0; + for (int i = 0; i < values.length; i++) + { + if (values[i]) + { + packedValue = (byte) (packedValue | (1 << i)); + } + } + + buffer.put(packedValue); + } + + /** + * This is used for writing longstrs. + * @param buffer + * @param data + */ + public static void writeLongstr(ByteBuffer buffer, byte[] data) + { + if (data != null) + { + writeUnsignedInteger(buffer, data.length); + buffer.put(data); + } + else + { + writeUnsignedInteger(buffer, 0); + } + } + + public static void writeTimestamp(ByteBuffer buffer, long timestamp) + { + writeUnsignedInteger(buffer, 0/*timestamp msb*/); + writeUnsignedInteger(buffer, timestamp); + } + + public static boolean[] readBooleans(ByteBuffer buffer) + { + byte packedValue = buffer.get(); + boolean[] result = new boolean[8]; + + for (int i = 0; i < 8; i++) + { + result[i] = ((packedValue & (1 << i)) != 0); + } + return result; + } + + public static FieldTable readFieldTable(ByteBuffer buffer) throws AMQFrameDecodingException + { + long length = buffer.getUnsignedInt(); + if (length == 0) + { + return null; + } + else + { + return new FieldTable(buffer, length); + } + } + + public static String readShortString(ByteBuffer buffer) + { + short length = buffer.getUnsigned(); + if (length == 0) + { + return null; + } + else + { + // this may seem rather odd to declare two array but testing has shown + // that constructing a string from a byte array is 5 (five) times slower + // than constructing one from a char array. + // this approach here is valid since we know that all the chars are + // ASCII (0-127) + byte[] stringBytes = new byte[length]; + buffer.get(stringBytes, 0, length); + char[] stringChars = new char[length]; + for (int i = 0; i < stringChars.length; i++) + { + stringChars[i] = (char) stringBytes[i]; + } + + return new String(stringChars); + } + } + + public static String readLongString(ByteBuffer buffer) + { + long length = buffer.getUnsignedInt(); + if (length == 0) + { + return null; + } + else + { + // this may seem rather odd to declare two array but testing has shown + // that constructing a string from a byte array is 5 (five) times slower + // than constructing one from a char array. + // this approach here is valid since we know that all the chars are + // ASCII (0-127) + byte[] stringBytes = new byte[(int)length]; + buffer.get(stringBytes, 0, (int)length); + char[] stringChars = new char[(int)length]; + for (int i = 0; i < stringChars.length; i++) + { + stringChars[i] = (char) stringBytes[i]; + } + return new String(stringChars); + } + } + + public static byte[] readLongstr(ByteBuffer buffer) throws AMQFrameDecodingException + { + long length = buffer.getUnsignedInt(); + if (length == 0) + { + return null; + } + else + { + byte[] result = new byte[(int)length]; + buffer.get(result); + return result; + } + } + + public static long readTimestamp(ByteBuffer buffer) + { + // Discard msb from AMQ timestamp + buffer.getUnsignedInt(); + return buffer.getUnsignedInt(); + } + + // Will barf with a NPE on a null input. Not sure whether it should return null or + // an empty field-table (which would be slower - perhaps unnecessarily). + // + // Some sample input and the result output: + // + // Input: "a=1" "a=2 c=3" "a=3 c=4 d" "a='four' b='five'" "a=bad" + // + // Parsing ... + // {a=1} + // Parsing ... + // {a=2, c=3} + // Parsing ... + // {a=3, c=4, d=null} + // Parsing ... + // {a=four, b=five} + // Parsing ... + // java.lang.IllegalArgumentException: a: Invalid integer in from . + // + public static FieldTable createFieldTableFromMessageSelector(String selector) + { + boolean debug = _logger.isDebugEnabled(); + + // TODO: Doesn't support embedded quotes properly. + String[] expressions = selector.split(" +"); + + FieldTable result = new FieldTable(); + + for (int i = 0; i < expressions.length; i++) + { + String expr = expressions[i]; + + if (debug) + { + _logger.debug("Expression = <" + expr + ">"); + } + + int equals = expr.indexOf('='); + + if (equals < 0) + { + // Existence check + result.put("S" + expr.trim(), null); + } + else + { + String key = expr.substring(0, equals).trim(); + String value = expr.substring(equals + 1).trim(); + + if (debug) + { + _logger.debug("Key = <" + key + ">, Value = <" + value + ">"); + } + + if (value.charAt(0) == '\'') + { + if (value.charAt(value.length() - 1) != '\'') + { + throw new IllegalArgumentException(key + ": Missing quote in <" + value + "> from <" + selector + ">."); + } + else + { + value = value.substring(1, value.length() - 1); + + result.put("S" + key, value); + } + } + else + { + try + { + int intValue = Integer.parseInt(value); + + result.put("i" + key, value); + } + catch (NumberFormatException e) + { + throw new IllegalArgumentException(key + ": Invalid integer in <" + value + "> from <" + selector + ">."); + + } + } + } + } + + if (debug) + { + _logger.debug("Field-table created from <" + selector + "> is <" + result + ">"); + } + + return (result); + + } + + static byte[] hexToByteArray(String id) + { + // Should check param for null, long enough for this check, upper-case and trailing char + String s = (id.charAt(1) == 'x') ? id.substring(2) : id; // strip 0x + + int len = s.length(); + int byte_len = len / 2; + byte[] b = new byte[byte_len]; + + for (int i = 0; i < byte_len; i++) + { + // fixme: refine these repetitive subscript calcs. + int ch = i * 2; + + byte b1 = Byte.parseByte(s.substring(ch, ch + 1), 16); + byte b2 = Byte.parseByte(s.substring(ch + 1, ch + 2), 16); + + b[i] = (byte) (b1 * 16 + b2); + } + + return (b); + } + + public static char[] convertToHexCharArray(byte[] from) + { + int length = from.length; + char[] result_buff = new char[length * 2 + 2]; + + result_buff[0] = '0'; + result_buff[1] = 'x'; + + int bite; + int dest = 2; + + for (int i = 0; i < length; i++) + { + bite = from[i]; + + if (bite < 0) + { + bite += 256; + } + + result_buff[dest++] = hex_chars[bite >> 4]; + result_buff[dest++] = hex_chars[bite & 0x0f]; + } + + return (result_buff); + } + + public static String convertToHexString(byte[] from) + { + return (new String(convertToHexCharArray(from))); + } + + public static String convertToHexString(ByteBuffer bb) + { + int size = bb.limit(); + + byte[] from = new byte[size]; + + // Is this not the same. + //bb.get(from, 0, size); + for (int i = 0; i < size; i++) + { + from[i] = bb.get(i); + } + + return (new String(convertToHexCharArray(from))); + } + + public static void main(String[] args) + { + for (int i = 0; i < args.length; i++) + { + String selector = args[i]; + + System.err.println("Parsing <" + selector + ">..."); + + try + { + System.err.println(createFieldTableFromMessageSelector(selector)); + } + catch (IllegalArgumentException e) + { + System.err.println(e); + } + } + } + + private static char hex_chars[] = {'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f'}; +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java b/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java new file mode 100644 index 0000000000..be456c8754 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java @@ -0,0 +1,322 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.log4j.Logger; +import org.apache.mina.common.ByteBuffer; + +import java.util.*; + +/** + * From the protocol document: + * field-table = short-integer *field-value-pair + * field-value-pair = field-name field-value + * field-name = short-string + * field-value = 'S' long-string + * / 'I' long-integer + * / 'D' decimal-value + * / 'T' long-integer + * decimal-value = decimals long-integer + * decimals = OCTET + */ +public class FieldTable extends LinkedHashMap +{ + private static final Logger _logger = Logger.getLogger(FieldTable.class); + private long _encodedSize = 0; + + public FieldTable() + { + super(); + } + + /** + * Construct a new field table. + * + * @param buffer the buffer from which to read data. The length byte must be read already + * @param length the length of the field table. Must be > 0. + * @throws AMQFrameDecodingException if there is an error decoding the table + */ + public FieldTable(ByteBuffer buffer, long length) throws AMQFrameDecodingException + { + super(); + final boolean debug = _logger.isDebugEnabled(); + assert length > 0; + _encodedSize = length; + int sizeRead = 0; + while (sizeRead < _encodedSize) + { + int sizeRemaining = buffer.remaining(); + final String key = EncodingUtils.readShortString(buffer); + // TODO: use proper charset decoder + byte iType = buffer.get(); + final char type = (char) iType; + Object value; + switch (type) + { + case 'S': + value = EncodingUtils.readLongString(buffer); + break; + case 'I': + value = new Long(buffer.getUnsignedInt()); + break; + default: + String msg = "Field '" + key + "' - unsupported field table type: " + type; + //some extra debug information... + msg += " (" + iType + "), length=" + length + ", sizeRead=" + sizeRead + ", sizeRemaining=" + sizeRemaining; + throw new AMQFrameDecodingException(msg); + } + sizeRead += (sizeRemaining - buffer.remaining()); + + if (debug) + { + _logger.debug("FieldTable::FieldTable(buffer," + length + "): Read type '" + type + "', key '" + key + "', value '" + value + "' (now read " + sizeRead + " of " + length + " encoded bytes)..."); + } + + // we deliberately want to call put in the parent class since we do + // not need to do the size calculations + super.put(key, value); + } + + if (debug) + { + _logger.debug("FieldTable::FieldTable(buffer," + length + "): Done."); + } + } + + public void writeToBuffer(ByteBuffer buffer) + { + final boolean debug = _logger.isDebugEnabled(); + + if (debug) + { + _logger.debug("FieldTable::writeToBuffer: Writing encoded size of " + _encodedSize + "..."); + } + + // write out the total length, which we have kept up to date as data is added + EncodingUtils.writeUnsignedInteger(buffer, _encodedSize); + final Iterator it = this.entrySet().iterator(); + while (it.hasNext()) + { + Map.Entry me = (Map.Entry) it.next(); + String key = (String) me.getKey(); + + EncodingUtils.writeShortStringBytes(buffer, key); + Object value = me.getValue(); + + if (debug) + { + _logger.debug("FieldTable::writeToBuffer: Writing key '" + key + "' of type " + value.getClass() + ", value '" + value + "'..."); + } + + if (value instanceof byte[]) + { + buffer.put((byte) 'S'); + EncodingUtils.writeLongstr(buffer, (byte[]) value); + } + else if (value instanceof String) + { + // TODO: look at using proper charset encoder + buffer.put((byte) 'S'); + EncodingUtils.writeLongStringBytes(buffer, (String) value); + } + else if (value instanceof Long) + { + // TODO: look at using proper charset encoder + buffer.put((byte) 'I'); + EncodingUtils.writeUnsignedInteger(buffer, ((Long) value).longValue()); + } + else + { + // Should never get here + throw new IllegalArgumentException("Key '" + key + "': Unsupported type in field table, type: " + ((value == null) ? "null-object" : value.getClass())); + } + } + + if (debug) + { + _logger.debug("FieldTable::writeToBuffer: Done."); + } + } + + public byte[] getDataAsBytes() + { + final ByteBuffer buffer = ByteBuffer.allocate((int) _encodedSize); // XXX: Is cast a problem? + final Iterator it = this.entrySet().iterator(); + while (it.hasNext()) + { + Map.Entry me = (Map.Entry) it.next(); + String key = (String) me.getKey(); + EncodingUtils.writeShortStringBytes(buffer, key); + Object value = me.getValue(); + if (value instanceof byte[]) + { + buffer.put((byte) 'S'); + EncodingUtils.writeLongstr(buffer, (byte[]) value); + } + else if (value instanceof String) + { + // TODO: look at using proper charset encoder + buffer.put((byte) 'S'); + EncodingUtils.writeLongStringBytes(buffer, (String) value); + } + else if (value instanceof char[]) + { + // TODO: look at using proper charset encoder + buffer.put((byte) 'S'); + EncodingUtils.writeLongStringBytes(buffer, (char[]) value); + } + else if (value instanceof Long || value instanceof Integer) + { + // TODO: look at using proper charset encoder + buffer.put((byte) 'I'); + EncodingUtils.writeUnsignedInteger(buffer, ((Long) value).longValue()); + } + else + { + // Should never get here + assert false; + } + } + final byte[] result = new byte[(int) _encodedSize]; + buffer.flip(); + buffer.get(result); + buffer.release(); + return result; + } + + public Object put(Object key, Object value) + { + final boolean debug = _logger.isDebugEnabled(); + + if (key == null) + { + throw new IllegalArgumentException("All keys must be Strings - was passed: null"); + } + else if (!(key instanceof String)) + { + throw new IllegalArgumentException("All keys must be Strings - was passed: " + key.getClass()); + } + + Object existing; + + if ((existing = super.remove(key)) != null) + { + if (debug) + { + _logger.debug("Found duplicate of key '" + key + "', previous value '" + existing + "' (" + existing.getClass() + "), to be replaced by '" + value + "', (" + value.getClass() + ") - stack trace of source of duplicate follows...", new Throwable().fillInStackTrace()); + } + + // If we are in effect deleting the value (see comment on null values being deleted + // below) then we also need to remove the name from the encoding length. + if (value == null) + { + _encodedSize -= EncodingUtils.encodedShortStringLength((String) key); + } + + // FIXME: Should be able to short-cut this process if the old and new values are + // the same object and/or type and size... + _encodedSize -= getEncodingSize(existing); + } + else + { + if (value != null) + { + _encodedSize += EncodingUtils.encodedShortStringLength((String) key); + } + } + + // For now: Setting a null value is the equivalent of deleting it. + // This is ambiguous in the JMS spec and needs thrashing out and potentially + // testing against other implementations. + if (value != null) + { + _encodedSize += getEncodingSize(value); + } + + return super.put(key, value); + } + + public Object remove(Object key) + { + if (super.containsKey(key)) + { + final Object value = super.remove(key); + _encodedSize -= EncodingUtils.encodedShortStringLength((String) key); + + // This check is, for now, unnecessary (we don't store null values). + if (value != null) + { + _encodedSize -= getEncodingSize(value); + } + + return value; + } + else + { + return null; + } + } + + /** + * @return unsigned integer + */ + public long getEncodedSize() + { + return _encodedSize; + } + + /** + * @return integer + */ + private static int getEncodingSize(Object value) + { + int encodingSize; + + // the extra byte if for the type indicator that is written out + if (value instanceof String) + { + encodingSize = 1 + EncodingUtils.encodedLongStringLength((String) value); + } + else if (value instanceof char[]) + { + encodingSize = 1 + EncodingUtils.encodedLongStringLength((char[]) value); + } + else if (value instanceof Integer) + { + encodingSize = 1 + 4; + } + else if (value instanceof Long) + { + encodingSize = 1 + 4; + } + else + { + throw new IllegalArgumentException("Unsupported type in field table: " + value.getClass()); + } + + return encodingSize; + } + + public Enumeration keys() + { + return new FieldTableKeyEnumeration(this); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/FieldTableKeyEnumeration.java b/java/common/src/main/java/org/apache/qpid/framing/FieldTableKeyEnumeration.java new file mode 100644 index 0000000000..e3ba9080c7 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/FieldTableKeyEnumeration.java @@ -0,0 +1,47 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import java.util.Enumeration; +import java.util.Iterator; + + +public class FieldTableKeyEnumeration implements Enumeration +{ + protected FieldTable _table; + protected Iterator _iterator; + + public FieldTableKeyEnumeration(FieldTable ft) + { + _table = ft; + _iterator = ft.keySet().iterator(); + } + + public boolean hasMoreElements() + { + return _iterator.hasNext(); + } + + public Object nextElement() + { + return _iterator.next(); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBody.java b/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBody.java new file mode 100644 index 0000000000..6bbdaaeed5 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBody.java @@ -0,0 +1,57 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.mina.common.ByteBuffer; + +public class HeartbeatBody extends AMQBody +{ + public static final byte TYPE = 8; + public static AMQFrame FRAME = new HeartbeatBody().toFrame(); + + protected byte getType() + { + return TYPE; + } + + protected int getSize() + { + return 0;//heartbeats we generate have no payload + } + + protected void writePayload(ByteBuffer buffer) + { + } + + protected void populateFromBuffer(ByteBuffer buffer, long size) throws AMQFrameDecodingException + { + if(size > 0) + { + //allow other implementations to have a payload, but ignore it: + buffer.skip((int) size); + } + } + + public AMQFrame toFrame() + { + return new AMQFrame(0, this); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBodyFactory.java b/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBodyFactory.java new file mode 100644 index 0000000000..97bd3d9253 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBodyFactory.java @@ -0,0 +1,31 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.mina.common.ByteBuffer; + +public class HeartbeatBodyFactory implements BodyFactory +{ + public AMQBody createBody(ByteBuffer in) throws AMQFrameDecodingException + { + return new HeartbeatBody(); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java b/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java new file mode 100644 index 0000000000..f0d5489527 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java @@ -0,0 +1,179 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.framing; + +import org.apache.mina.common.ByteBuffer; +import org.apache.mina.common.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.qpid.AMQException; + +public class ProtocolInitiation extends AMQDataBlock implements EncodableAMQDataBlock +{ + public char[] header = new char[]{'A','M','Q','P'}; + // TODO: generate these constants automatically from the xml protocol spec file + + private static byte CURRENT_PROTOCOL_CLASS = 1; + private static final int CURRENT_PROTOCOL_INSTANCE = 1; + + public byte protocolClass = CURRENT_PROTOCOL_CLASS; + public byte protocolInstance = CURRENT_PROTOCOL_INSTANCE; + public byte protocolMajor; + public byte protocolMinor; + +// public ProtocolInitiation() {} + + public ProtocolInitiation(byte major, byte minor) + { + protocolMajor = major; + protocolMinor = minor; + } + + public long getSize() + { + return 4 + 1 + 1 + 1 + 1; + } + + public void writePayload(ByteBuffer buffer) + { + for (int i = 0; i < header.length; i++) + { + buffer.put((byte) header[i]); + } + buffer.put(protocolClass); + buffer.put(protocolInstance); + buffer.put(protocolMajor); + buffer.put(protocolMinor); + } + + public void populateFromBuffer(ByteBuffer buffer) throws AMQException + { + throw new AMQException("Method not implemented"); + } + + public boolean equals(Object o) + { + if (!(o instanceof ProtocolInitiation)) + { + return false; + } + + ProtocolInitiation pi = (ProtocolInitiation) o; + if (pi.header == null) + { + return false; + } + + if (header.length != pi.header.length) + { + return false; + } + + for (int i = 0; i < header.length; i++) + { + if (header[i] != pi.header[i]) + { + return false; + } + } + + return (protocolClass == pi.protocolClass && + protocolInstance == pi.protocolInstance && + protocolMajor == pi.protocolMajor && + protocolMinor == pi.protocolMinor); + } + + public static class Decoder //implements MessageDecoder + { + /** + * + * @param session + * @param in + * @return true if we have enough data to decode the PI frame fully, false if more + * data is required + */ + public boolean decodable(IoSession session, ByteBuffer in) + { + return (in.remaining() >= 8); + } + + public void decode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) + throws Exception + { + byte[] theHeader = new byte[4]; + in.get(theHeader); + ProtocolInitiation pi = new ProtocolInitiation((byte)0, (byte)0); + pi.header = new char[]{(char) theHeader[0],(char) theHeader[CURRENT_PROTOCOL_INSTANCE],(char) theHeader[2], (char) theHeader[3]}; + String stringHeader = new String(pi.header); + if (!"AMQP".equals(stringHeader)) + { + throw new AMQProtocolHeaderException("Invalid protocol header - read " + stringHeader); + } + pi.protocolClass = in.get(); + pi.protocolInstance = in.get(); + pi.protocolMajor = in.get(); + pi.protocolMinor = in.get(); + out.write(pi); + } + } + + public void checkVersion(ProtocolVersionList pvl) throws AMQException + { + if (protocolClass != CURRENT_PROTOCOL_CLASS) + { + throw new AMQProtocolClassException("Protocol class " + CURRENT_PROTOCOL_CLASS + " was expected; received " + + protocolClass); + } + if (protocolInstance != CURRENT_PROTOCOL_INSTANCE) + { + throw new AMQProtocolInstanceException("Protocol instance " + CURRENT_PROTOCOL_INSTANCE + " was expected; received " + + protocolInstance); + } + /* + if (protocolMajor != CURRENT_PROTOCOL_VERSION_MAJOR) + { + throw new AMQProtocolVersionException("Protocol major version " + CURRENT_PROTOCOL_VERSION_MAJOR + + " was expected; received " + protocolMajor); + } + if (protocolMinor != CURRENT_PROTOCOL_VERSION_MINOR) + { + throw new AMQProtocolVersionException("Protocol minor version " + CURRENT_PROTOCOL_VERSION_MINOR + + " was expected; received " + protocolMinor); + } + */ + + /* Look through list of available protocol versions */ + boolean found = false; + for (int i=0; i _eventQueue = new ConcurrentLinkedQueue(); + private final AtomicBoolean _active = new AtomicBoolean(); + private final AtomicInteger _refCount = new AtomicInteger(); + private final JobCompletionHandler _completionHandler; + + Job(IoSession session, JobCompletionHandler completionHandler, int maxEvents) + { + _session = session; + _completionHandler = completionHandler; + _maxEvents = maxEvents; + } + + void acquire() + { + _refCount.incrementAndGet(); + } + + void release() + { + _refCount.decrementAndGet(); + } + + boolean isReferenced() + { + return _refCount.get() > 0; + } + + void add(Event evt) + { + _eventQueue.add(evt); + } + + void processAll() + { + //limit the number of events processed in one run + for (int i = 0; i < _maxEvents; i++) + { + Event e = _eventQueue.poll(); + if (e == null) + { + break; + } + else + { + e.process(_session); + } + } + } + + boolean isComplete() + { + return _eventQueue.peek() == null; + } + + boolean activate() + { + return _active.compareAndSet(false, true); + } + + void deactivate() + { + _active.set(false); + } + + public void run() + { + processAll(); + deactivate(); + _completionHandler.completed(_session, this); + } + + + static interface JobCompletionHandler + { + public void completed(IoSession session, Job job); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java b/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java new file mode 100644 index 0000000000..ba614a26c2 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java @@ -0,0 +1,186 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.pool; + +import org.apache.log4j.Logger; +import org.apache.mina.common.IdleStatus; +import org.apache.mina.common.IoFilterAdapter; +import org.apache.mina.common.IoSession; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +public class PoolingFilter extends IoFilterAdapter implements Job.JobCompletionHandler +{ + private static final Logger _logger = Logger.getLogger(PoolingFilter.class); + public static final Set READ_EVENTS = new HashSet(Arrays.asList(EventType.RECEIVED)); + public static final Set WRITE_EVENTS = new HashSet(Arrays.asList(EventType.WRITE)); + + private final ConcurrentMap _jobs = new ConcurrentHashMap(); + private final ReferenceCountingExecutorService _poolReference; + private final Set _asyncTypes; + + private final String _name; + private final int _maxEvents = Integer.getInteger("amqj.server.read_write_pool.max_events", 10); + + public PoolingFilter(ReferenceCountingExecutorService refCountingPool, Set asyncTypes, String name) + { + _poolReference = refCountingPool; + _asyncTypes = asyncTypes; + _name = name; + } + + private void fireEvent(IoSession session, Event event) + { + if (_asyncTypes.contains(event.getType())) + { + Job job = getJobForSession(session); + job.acquire(); //prevents this job being removed from _jobs + job.add(event); + if (job.activate()) + { + _poolReference.getPool().execute(job); + } + } + else + { + event.process(session); + } + } + + private Job getJobForSession(IoSession session) + { + Job job = _jobs.get(session); + return job == null ? createJobForSession(session) : job; + } + + private Job createJobForSession(IoSession session) + { + return addJobForSession(session, new Job(session, this, _maxEvents)); + } + + private Job addJobForSession(IoSession session, Job job) + { + //atomic so ensures all threads agree on the same job + Job existing = _jobs.putIfAbsent(session, job); + return existing == null ? job : existing; + } + + //Job.JobCompletionHandler + public void completed(IoSession session, Job job) + { + if (job.isComplete()) + { + job.release(); + if (!job.isReferenced()) + { + _jobs.remove(session); + } + } + else + { + if (job.activate()) + { + _poolReference.getPool().execute(job); + } + } + } + + //IoFilter methods that are processed by threads on the pool + + public void sessionOpened(NextFilter nextFilter, IoSession session) throws Exception + { + fireEvent(session, new Event(nextFilter, EventType.OPENED, null)); + } + + public void sessionClosed(NextFilter nextFilter, IoSession session) throws Exception + { + fireEvent(session, new Event(nextFilter, EventType.CLOSED, null)); + } + + public void sessionIdle(NextFilter nextFilter, IoSession session, + IdleStatus status) throws Exception + { + fireEvent(session, new Event(nextFilter, EventType.IDLE, status)); + } + + public void exceptionCaught(NextFilter nextFilter, IoSession session, + Throwable cause) throws Exception + { + fireEvent(session, new Event(nextFilter, EventType.EXCEPTION, cause)); + } + + public void messageReceived(NextFilter nextFilter, IoSession session, + Object message) throws Exception + { + //ByteBufferUtil.acquireIfPossible( message ); + fireEvent(session, new Event(nextFilter, EventType.RECEIVED, message)); + } + + public void messageSent(NextFilter nextFilter, IoSession session, + Object message) throws Exception + { + //ByteBufferUtil.acquireIfPossible( message ); + fireEvent(session, new Event(nextFilter, EventType.SENT, message)); + } + + public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception + { + fireEvent(session, new Event(nextFilter, EventType.WRITE, writeRequest)); + } + + //IoFilter methods that are processed on current thread (NOT on pooled thread) + + public void filterClose(NextFilter nextFilter, IoSession session) throws Exception + { + nextFilter.filterClose(session); + } + + public void sessionCreated(NextFilter nextFilter, IoSession session) + { + nextFilter.sessionCreated(session); + } + + public String toString() + { + return _name; + } + + // LifeCycle methods + + public void init() + { + _logger.info("Init called on PoolingFilter " + toString()); + // called when the filter is initialised in the chain. If the reference count is + // zero this acquire will initialise the pool + _poolReference.acquireExecutorService(); + } + + public void destroy() + { + _logger.info("Destroy called on PoolingFilter " + toString()); + // when the reference count gets to zero we release the executor service + _poolReference.releaseExecutorService(); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/pool/ReadWriteThreadModel.java b/java/common/src/main/java/org/apache/qpid/pool/ReadWriteThreadModel.java new file mode 100644 index 0000000000..d4dbf1309a --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/pool/ReadWriteThreadModel.java @@ -0,0 +1,40 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.pool; + +import org.apache.mina.common.IoFilterChain; +import org.apache.mina.filter.ReferenceCountingIoFilter; +import org.apache.mina.common.ThreadModel; + +public class ReadWriteThreadModel implements ThreadModel +{ + public void buildFilterChain(IoFilterChain chain) throws Exception + { + ReferenceCountingExecutorService executor = ReferenceCountingExecutorService.getInstance(); + PoolingFilter asyncRead = new PoolingFilter(executor, PoolingFilter.READ_EVENTS, + "AsynchronousReadFilter"); + PoolingFilter asyncWrite = new PoolingFilter(executor, PoolingFilter.WRITE_EVENTS, + "AsynchronousWriteFilter"); + + chain.addFirst("AsynchronousReadFilter", new ReferenceCountingIoFilter(asyncRead)); + chain.addLast("AsynchronousWriteFilter", new ReferenceCountingIoFilter(asyncWrite)); + } +} diff --git a/java/common/src/main/java/org/apache/qpid/pool/ReferenceCountingExecutorService.java b/java/common/src/main/java/org/apache/qpid/pool/ReferenceCountingExecutorService.java new file mode 100644 index 0000000000..637464f247 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/pool/ReferenceCountingExecutorService.java @@ -0,0 +1,98 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.pool; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * We share the executor service among several PoolingFilters. This class reference counts + * how many filter chains are using the executor service and destroys the service, thus + * freeing up its threads, when the count reaches zero. It recreates the service when + * the count is incremented. + * + * This is particularly important on the client where failing to destroy the executor + * service prevents the JVM from shutting down due to the existence of non-daemon threads. + * + */ +public class ReferenceCountingExecutorService +{ + private static final int MINIMUM_POOL_SIZE = 4; + private static final int NUM_CPUS = Runtime.getRuntime().availableProcessors(); + private static final int DEFAULT_POOL_SIZE = Math.max(NUM_CPUS, MINIMUM_POOL_SIZE); + + /** + * We need to be able to check the current reference count and if necessary + * create the executor service atomically. + */ + private static final ReferenceCountingExecutorService _instance = new ReferenceCountingExecutorService(); + + private final Object _lock = new Object(); + + private ExecutorService _pool; + + private int _refCount = 0; + + private int _poolSize = Integer.getInteger("amqj.read_write_pool_size", DEFAULT_POOL_SIZE); + + public static ReferenceCountingExecutorService getInstance() + { + return _instance; + } + + private ReferenceCountingExecutorService() + { + } + + ExecutorService acquireExecutorService() + { + synchronized (_lock) + { + if (_refCount++ == 0) + { + _pool = Executors.newFixedThreadPool(_poolSize); + } + return _pool; + } + } + + void releaseExecutorService() + { + synchronized (_lock) + { + if (--_refCount == 0) + { + _pool.shutdownNow(); + } + } + } + + /** + * The filters that use the executor service should call this method to get access + * to the service. Note that this method does not alter the reference count. + * + * @return the underlying executor service + */ + public ExecutorService getPool() + { + return _pool; + } +} diff --git a/java/common/src/main/java/org/apache/qpid/protocol/AMQConstant.java b/java/common/src/main/java/org/apache/qpid/protocol/AMQConstant.java new file mode 100644 index 0000000000..fc83c0726d --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/protocol/AMQConstant.java @@ -0,0 +1,108 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.protocol; + +import java.util.Map; +import java.util.HashMap; + +public final class AMQConstant +{ + private int _code; + + private String _name; + + private static Map _codeMap = new HashMap(); + + private AMQConstant(int code, String name, boolean map) + { + _code = code; + _name = name; + if (map) + { + _codeMap.put(new Integer(code), this); + } + } + + public String toString() + { + return _code + ": " + _name; + } + + public int getCode() + { + return _code; + } + + public String getName() + { + return _name; + } + + public static final AMQConstant FRAME_MIN_SIZE = new AMQConstant(4096, "frame min size", true); + + public static final AMQConstant FRAME_END = new AMQConstant(206, "frame end", true); + + public static final AMQConstant REPLY_SUCCESS = new AMQConstant(200, "reply success", true); + + public static final AMQConstant NOT_DELIVERED = new AMQConstant(310, "not delivered", true); + + public static final AMQConstant MESSAGE_TOO_LARGE = new AMQConstant(311, "message too large", true); + + public static final AMQConstant NO_ROUTE = new AMQConstant(312, "no route", true); + + public static final AMQConstant NO_CONSUMERS = new AMQConstant(313, "no consumers", true); + + public static final AMQConstant CONTEXT_IN_USE = new AMQConstant(320, "context in use", true); + + public static final AMQConstant CONTEXT_UNKNOWN = new AMQConstant(321, "context unknown", true); + + public static final AMQConstant INVALID_PATH = new AMQConstant(402, "invalid path", true); + + public static final AMQConstant ACCESS_REFUSED = new AMQConstant(403, "access refused", true); + + public static final AMQConstant NOT_FOUND = new AMQConstant(404, "not found", true); + + public static final AMQConstant FRAME_ERROR = new AMQConstant(501, "frame error", true); + + public static final AMQConstant SYNTAX_ERROR = new AMQConstant(502, "syntax error", true); + + public static final AMQConstant COMMAND_INVALID = new AMQConstant(503, "command invalid", true); + + public static final AMQConstant CHANNEL_ERROR = new AMQConstant(504, "channel error", true); + + public static final AMQConstant RESOURCE_ERROR = new AMQConstant(506, "resource error", true); + + public static final AMQConstant NOT_ALLOWED = new AMQConstant(507, "not allowed", true); + + public static final AMQConstant NOT_IMPLEMENTED = new AMQConstant(540, "not implemented", true); + + public static final AMQConstant INTERNAL_ERROR = new AMQConstant(541, "internal error", true); + + public static AMQConstant getConstant(int code) + { + AMQConstant c = (AMQConstant) _codeMap.get(new Integer(code)); + if (c == null) + { + c = new AMQConstant(code, "unknown code", false); + } + return c; + } +} diff --git a/java/common/src/main/java/org/apache/qpid/ssl/BogusSSLContextFactory.java b/java/common/src/main/java/org/apache/qpid/ssl/BogusSSLContextFactory.java new file mode 100644 index 0000000000..fee02c9d93 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/ssl/BogusSSLContextFactory.java @@ -0,0 +1,159 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.ssl; + +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import java.io.IOException; +import java.io.InputStream; +import java.security.GeneralSecurityException; +import java.security.KeyStore; + +/** + * Factory to create a bogus SSLContext. This means that it is easy to test SSL but this + * cannot be used in a production environment. + *

+ * This is based on the sample that comes with MINA, written by Trustin Lee + */ +public class BogusSSLContextFactory +{ + /** + * Protocol to use. + */ + private static final String PROTOCOL = "TLS"; + + /** + * Bougus Server certificate keystore file name. + */ + private static final String BOGUS_KEYSTORE = "qpid.cert"; + + // NOTE: The keystore was generated using keytool: + // keytool -genkey -alias qpid -keysize 512 -validity 3650 + // -keyalg RSA -dname "CN=amqp.org" -keypass qpidpw + // -storepass qpidpw -keystore qpid.cert + + private static final char[] BOGUS_KEYSTORE_PASSWORD = {'q', 'p', 'i', 'd', 'p', 'w'}; + + private static SSLContext serverInstance = null; + + private static SSLContext clientInstance = null; + + /** + * Get SSLContext singleton. + * + * @return SSLContext + * @throws java.security.GeneralSecurityException + */ + public static SSLContext getInstance(boolean server) + throws GeneralSecurityException + { + SSLContext retInstance; + if (server) + { + // FIXME: looks like double-checking locking + if (serverInstance == null) + { + synchronized (BogusSSLContextFactory.class) + { + if (serverInstance == null) + { + try + { + serverInstance = createBougusServerSSLContext(); + } + catch (Exception ioe) + { + throw new GeneralSecurityException( + "Can't create Server SSLContext:" + ioe); + } + } + } + } + retInstance = serverInstance; + } + else + { + // FIXME: looks like double-checking locking + if (clientInstance == null) + { + synchronized (BogusSSLContextFactory.class) + { + if (clientInstance == null) + { + clientInstance = createBougusClientSSLContext(); + } + } + } + retInstance = clientInstance; + } + return retInstance; + } + + private static SSLContext createBougusServerSSLContext() + throws GeneralSecurityException, IOException + { + // Create keystore + KeyStore ks = KeyStore.getInstance("JKS"); + InputStream in = null; + try + { + in = BogusSSLContextFactory.class.getResourceAsStream(BOGUS_KEYSTORE); + if (in == null) + { + throw new IOException("Unable to load keystore resource: " + BOGUS_KEYSTORE); + } + ks.load(in, BOGUS_KEYSTORE_PASSWORD); + } + finally + { + if (in != null) + { + //noinspection EmptyCatchBlock + try + { + in.close(); + } + catch (IOException ignored) + { + } + } + } + + // Set up key manager factory to use our key store + KeyManagerFactory kmf = KeyManagerFactory.getInstance("SunX509"); + kmf.init(ks, BOGUS_KEYSTORE_PASSWORD); + + // Initialize the SSLContext to work with our key managers. + SSLContext sslContext = SSLContext.getInstance(PROTOCOL); + sslContext.init(kmf.getKeyManagers(), BogusTrustManagerFactory.X509_MANAGERS, null); + + return sslContext; + } + + private static SSLContext createBougusClientSSLContext() + throws GeneralSecurityException + { + SSLContext context = SSLContext.getInstance(PROTOCOL); + context.init(null, BogusTrustManagerFactory.X509_MANAGERS, null); + return context; + } + +} diff --git a/java/common/src/main/java/org/apache/qpid/ssl/BogusTrustManagerFactory.java b/java/common/src/main/java/org/apache/qpid/ssl/BogusTrustManagerFactory.java new file mode 100644 index 0000000000..4fb6f75b8d --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/ssl/BogusTrustManagerFactory.java @@ -0,0 +1,82 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.ssl; + +import javax.net.ssl.ManagerFactoryParameters; +import javax.net.ssl.TrustManager; +import javax.net.ssl.TrustManagerFactorySpi; +import javax.net.ssl.X509TrustManager; +import java.security.InvalidAlgorithmParameterException; +import java.security.KeyStore; +import java.security.KeyStoreException; +import java.security.cert.CertificateException; +import java.security.cert.X509Certificate; + +/** + * Bogus trust manager factory. Used to make testing SSL simpler - i.e no need to + * mess about with keystores. + *

+ * This is based on the example that comes with MINA, written by Trustin Lee. + */ +class BogusTrustManagerFactory extends TrustManagerFactorySpi +{ + + static final X509TrustManager X509 = new X509TrustManager() + { + public void checkClientTrusted(X509Certificate[] x509Certificates, + String s) throws CertificateException + { + } + + public void checkServerTrusted(X509Certificate[] x509Certificates, + String s) throws CertificateException + { + } + + public X509Certificate[] getAcceptedIssuers() + { + return new X509Certificate[ 0 ]; + } + }; + + static final TrustManager[] X509_MANAGERS = new TrustManager[]{X509}; + + public BogusTrustManagerFactory() + { + } + + protected TrustManager[] engineGetTrustManagers() + { + return X509_MANAGERS; + } + + protected void engineInit(KeyStore keystore) throws KeyStoreException + { + // noop + } + + protected void engineInit( + ManagerFactoryParameters managerFactoryParameters) + throws InvalidAlgorithmParameterException + { + // noop + } +} diff --git a/java/common/src/main/java/org/apache/qpid/ssl/SSLServerSocketFactory.java b/java/common/src/main/java/org/apache/qpid/ssl/SSLServerSocketFactory.java new file mode 100644 index 0000000000..90b3589752 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/ssl/SSLServerSocketFactory.java @@ -0,0 +1,108 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.ssl; + +import javax.net.ServerSocketFactory; +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.security.GeneralSecurityException; + +/** + * Simple Server Socket factory to create sockets with or without SSL enabled. + * If SSL enabled a "bogus" SSL Context is used (suitable for test purposes) + *

+ * This is based on the example that comes with MINA, written by Trustin Lee. + */ +public class SSLServerSocketFactory extends javax.net.ServerSocketFactory +{ + private static boolean sslEnabled = false; + + private static javax.net.ServerSocketFactory sslFactory = null; + + private static ServerSocketFactory factory = null; + + public SSLServerSocketFactory() + { + super(); + } + + public ServerSocket createServerSocket(int port) throws IOException + { + return new ServerSocket(port); + } + + public ServerSocket createServerSocket(int port, int backlog) + throws IOException + { + return new ServerSocket(port, backlog); + } + + public ServerSocket createServerSocket(int port, int backlog, + InetAddress ifAddress) + throws IOException + { + return new ServerSocket(port, backlog, ifAddress); + } + + public static javax.net.ServerSocketFactory getServerSocketFactory() + throws IOException + { + if (isSslEnabled()) + { + if (sslFactory == null) + { + try + { + sslFactory = BogusSSLContextFactory.getInstance(true) + .getServerSocketFactory(); + } + catch (GeneralSecurityException e) + { + IOException ioe = new IOException( + "could not create SSL socket"); + ioe.initCause(e); + throw ioe; + } + } + return sslFactory; + } + else + { + if (factory == null) + { + factory = new SSLServerSocketFactory(); + } + return factory; + } + + } + + public static boolean isSslEnabled() + { + return sslEnabled; + } + + public static void setSslEnabled(boolean newSslEnabled) + { + sslEnabled = newSslEnabled; + } +} diff --git a/java/common/src/main/java/org/apache/qpid/ssl/SSLSocketFactory.java b/java/common/src/main/java/org/apache/qpid/ssl/SSLSocketFactory.java new file mode 100644 index 0000000000..ef9820f067 --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/ssl/SSLSocketFactory.java @@ -0,0 +1,138 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.ssl; + +import javax.net.SocketFactory; +import java.io.IOException; +import java.net.InetAddress; +import java.net.Socket; +import java.net.UnknownHostException; +import java.security.GeneralSecurityException; + +/** + * Simple Socket factory to create sockets with or without SSL enabled. + * If SSL enabled a "bogus" SSL Context is used (suitable for test purposes). + *

+ * This is based on an example that comes with MINA, written by Trustin Lee. + */ +public class SSLSocketFactory extends SocketFactory +{ + private static boolean sslEnabled = false; + + private static javax.net.ssl.SSLSocketFactory sslFactory = null; + + private static javax.net.SocketFactory factory = null; + + public SSLSocketFactory() + { + super(); + } + + public Socket createSocket(String arg1, int arg2) throws IOException, + UnknownHostException + { + if (isSslEnabled()) + { + return getSSLFactory().createSocket(arg1, arg2); + } + else + { + return new Socket(arg1, arg2); + } + } + + public Socket createSocket(String arg1, int arg2, InetAddress arg3, + int arg4) throws IOException, + UnknownHostException + { + if (isSslEnabled()) + { + return getSSLFactory().createSocket(arg1, arg2, arg3, arg4); + } + else + { + return new Socket(arg1, arg2, arg3, arg4); + } + } + + public Socket createSocket(InetAddress arg1, int arg2) + throws IOException + { + if (isSslEnabled()) + { + return getSSLFactory().createSocket(arg1, arg2); + } + else + { + return new Socket(arg1, arg2); + } + } + + public Socket createSocket(InetAddress arg1, int arg2, InetAddress arg3, + int arg4) throws IOException + { + if (isSslEnabled()) + { + return getSSLFactory().createSocket(arg1, arg2, arg3, arg4); + } + else + { + return new Socket(arg1, arg2, arg3, arg4); + } + } + + public static javax.net.SocketFactory getSocketFactory() + { + if (factory == null) + { + factory = new SSLSocketFactory(); + } + return factory; + } + + private javax.net.ssl.SSLSocketFactory getSSLFactory() + { + if (sslFactory == null) + { + try + { + sslFactory = BogusSSLContextFactory.getInstance(false) + .getSocketFactory(); + } + catch (GeneralSecurityException e) + { + throw new RuntimeException("could not create SSL socket", e); + } + } + return sslFactory; + } + + public static boolean isSslEnabled() + { + return sslEnabled; + } + + public static void setSslEnabled(boolean newSslEnabled) + { + sslEnabled = newSslEnabled; + } + +} diff --git a/java/common/src/main/java/org/apache/qpid/url/AMQBindingURL.java b/java/common/src/main/java/org/apache/qpid/url/AMQBindingURL.java new file mode 100644 index 0000000000..b6a0bd500a --- /dev/null +++ b/java/common/src/main/java/org/apache/qpid/url/AMQBindingURL.java @@ -0,0 +1,263 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.url; + +import org.apache.qpid.url.BindingURL; +import org.apache.qpid.url.URLHelper; +import org.apache.qpid.exchange.ExchangeDefaults; + +import java.util.HashMap; +import java.net.URI; +import java.net.URISyntaxException; + +public class AMQBindingURL implements BindingURL +{ + String _url; + String _exchangeClass; + String _exchangeName; + String _destinationName; + String _queueName; + private HashMap _options; + + + public AMQBindingURL(String url) throws URLSyntaxException + { + //format: + // :///[]/[]?

- * This is based on the sample that comes with MINA, written by Trustin Lee - */ -public class BogusSSLContextFactory -{ - /** - * Protocol to use. - */ - private static final String PROTOCOL = "TLS"; - - /** - * Bougus Server certificate keystore file name. - */ - private static final String BOGUS_KEYSTORE = "qpid.cert"; - - // NOTE: The keystore was generated using keytool: - // keytool -genkey -alias qpid -keysize 512 -validity 3650 - // -keyalg RSA -dname "CN=amqp.org" -keypass qpidpw - // -storepass qpidpw -keystore qpid.cert - - private static final char[] BOGUS_KEYSTORE_PASSWORD = {'q', 'p', 'i', 'd', 'p', 'w'}; - - private static SSLContext serverInstance = null; - - private static SSLContext clientInstance = null; - - /** - * Get SSLContext singleton. - * - * @return SSLContext - * @throws java.security.GeneralSecurityException - */ - public static SSLContext getInstance(boolean server) - throws GeneralSecurityException - { - SSLContext retInstance; - if (server) - { - // FIXME: looks like double-checking locking - if (serverInstance == null) - { - synchronized (BogusSSLContextFactory.class) - { - if (serverInstance == null) - { - try - { - serverInstance = createBougusServerSSLContext(); - } - catch (Exception ioe) - { - throw new GeneralSecurityException( - "Can't create Server SSLContext:" + ioe); - } - } - } - } - retInstance = serverInstance; - } - else - { - // FIXME: looks like double-checking locking - if (clientInstance == null) - { - synchronized (BogusSSLContextFactory.class) - { - if (clientInstance == null) - { - clientInstance = createBougusClientSSLContext(); - } - } - } - retInstance = clientInstance; - } - return retInstance; - } - - private static SSLContext createBougusServerSSLContext() - throws GeneralSecurityException, IOException - { - // Create keystore - KeyStore ks = KeyStore.getInstance("JKS"); - InputStream in = null; - try - { - in = BogusSSLContextFactory.class.getResourceAsStream(BOGUS_KEYSTORE); - if (in == null) - { - throw new IOException("Unable to load keystore resource: " + BOGUS_KEYSTORE); - } - ks.load(in, BOGUS_KEYSTORE_PASSWORD); - } - finally - { - if (in != null) - { - //noinspection EmptyCatchBlock - try - { - in.close(); - } - catch (IOException ignored) - { - } - } - } - - // Set up key manager factory to use our key store - KeyManagerFactory kmf = KeyManagerFactory.getInstance("SunX509"); - kmf.init(ks, BOGUS_KEYSTORE_PASSWORD); - - // Initialize the SSLContext to work with our key managers. - SSLContext sslContext = SSLContext.getInstance(PROTOCOL); - sslContext.init(kmf.getKeyManagers(), BogusTrustManagerFactory.X509_MANAGERS, null); - - return sslContext; - } - - private static SSLContext createBougusClientSSLContext() - throws GeneralSecurityException - { - SSLContext context = SSLContext.getInstance(PROTOCOL); - context.init(null, BogusTrustManagerFactory.X509_MANAGERS, null); - return context; - } - -} diff --git a/java/common/src/org/apache/qpid/ssl/BogusTrustManagerFactory.java b/java/common/src/org/apache/qpid/ssl/BogusTrustManagerFactory.java deleted file mode 100644 index 4fb6f75b8d..0000000000 --- a/java/common/src/org/apache/qpid/ssl/BogusTrustManagerFactory.java +++ /dev/null @@ -1,82 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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.ssl; - -import javax.net.ssl.ManagerFactoryParameters; -import javax.net.ssl.TrustManager; -import javax.net.ssl.TrustManagerFactorySpi; -import javax.net.ssl.X509TrustManager; -import java.security.InvalidAlgorithmParameterException; -import java.security.KeyStore; -import java.security.KeyStoreException; -import java.security.cert.CertificateException; -import java.security.cert.X509Certificate; - -/** - * Bogus trust manager factory. Used to make testing SSL simpler - i.e no need to - * mess about with keystores. - *

- * This is based on the example that comes with MINA, written by Trustin Lee. - */ -class BogusTrustManagerFactory extends TrustManagerFactorySpi -{ - - static final X509TrustManager X509 = new X509TrustManager() - { - public void checkClientTrusted(X509Certificate[] x509Certificates, - String s) throws CertificateException - { - } - - public void checkServerTrusted(X509Certificate[] x509Certificates, - String s) throws CertificateException - { - } - - public X509Certificate[] getAcceptedIssuers() - { - return new X509Certificate[ 0 ]; - } - }; - - static final TrustManager[] X509_MANAGERS = new TrustManager[]{X509}; - - public BogusTrustManagerFactory() - { - } - - protected TrustManager[] engineGetTrustManagers() - { - return X509_MANAGERS; - } - - protected void engineInit(KeyStore keystore) throws KeyStoreException - { - // noop - } - - protected void engineInit( - ManagerFactoryParameters managerFactoryParameters) - throws InvalidAlgorithmParameterException - { - // noop - } -} diff --git a/java/common/src/org/apache/qpid/ssl/SSLServerSocketFactory.java b/java/common/src/org/apache/qpid/ssl/SSLServerSocketFactory.java deleted file mode 100644 index 90b3589752..0000000000 --- a/java/common/src/org/apache/qpid/ssl/SSLServerSocketFactory.java +++ /dev/null @@ -1,108 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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.ssl; - -import javax.net.ServerSocketFactory; -import java.io.IOException; -import java.net.InetAddress; -import java.net.ServerSocket; -import java.security.GeneralSecurityException; - -/** - * Simple Server Socket factory to create sockets with or without SSL enabled. - * If SSL enabled a "bogus" SSL Context is used (suitable for test purposes) - *

- * This is based on the example that comes with MINA, written by Trustin Lee. - */ -public class SSLServerSocketFactory extends javax.net.ServerSocketFactory -{ - private static boolean sslEnabled = false; - - private static javax.net.ServerSocketFactory sslFactory = null; - - private static ServerSocketFactory factory = null; - - public SSLServerSocketFactory() - { - super(); - } - - public ServerSocket createServerSocket(int port) throws IOException - { - return new ServerSocket(port); - } - - public ServerSocket createServerSocket(int port, int backlog) - throws IOException - { - return new ServerSocket(port, backlog); - } - - public ServerSocket createServerSocket(int port, int backlog, - InetAddress ifAddress) - throws IOException - { - return new ServerSocket(port, backlog, ifAddress); - } - - public static javax.net.ServerSocketFactory getServerSocketFactory() - throws IOException - { - if (isSslEnabled()) - { - if (sslFactory == null) - { - try - { - sslFactory = BogusSSLContextFactory.getInstance(true) - .getServerSocketFactory(); - } - catch (GeneralSecurityException e) - { - IOException ioe = new IOException( - "could not create SSL socket"); - ioe.initCause(e); - throw ioe; - } - } - return sslFactory; - } - else - { - if (factory == null) - { - factory = new SSLServerSocketFactory(); - } - return factory; - } - - } - - public static boolean isSslEnabled() - { - return sslEnabled; - } - - public static void setSslEnabled(boolean newSslEnabled) - { - sslEnabled = newSslEnabled; - } -} diff --git a/java/common/src/org/apache/qpid/ssl/SSLSocketFactory.java b/java/common/src/org/apache/qpid/ssl/SSLSocketFactory.java deleted file mode 100644 index ef9820f067..0000000000 --- a/java/common/src/org/apache/qpid/ssl/SSLSocketFactory.java +++ /dev/null @@ -1,138 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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.ssl; - -import javax.net.SocketFactory; -import java.io.IOException; -import java.net.InetAddress; -import java.net.Socket; -import java.net.UnknownHostException; -import java.security.GeneralSecurityException; - -/** - * Simple Socket factory to create sockets with or without SSL enabled. - * If SSL enabled a "bogus" SSL Context is used (suitable for test purposes). - *

- * This is based on an example that comes with MINA, written by Trustin Lee. - */ -public class SSLSocketFactory extends SocketFactory -{ - private static boolean sslEnabled = false; - - private static javax.net.ssl.SSLSocketFactory sslFactory = null; - - private static javax.net.SocketFactory factory = null; - - public SSLSocketFactory() - { - super(); - } - - public Socket createSocket(String arg1, int arg2) throws IOException, - UnknownHostException - { - if (isSslEnabled()) - { - return getSSLFactory().createSocket(arg1, arg2); - } - else - { - return new Socket(arg1, arg2); - } - } - - public Socket createSocket(String arg1, int arg2, InetAddress arg3, - int arg4) throws IOException, - UnknownHostException - { - if (isSslEnabled()) - { - return getSSLFactory().createSocket(arg1, arg2, arg3, arg4); - } - else - { - return new Socket(arg1, arg2, arg3, arg4); - } - } - - public Socket createSocket(InetAddress arg1, int arg2) - throws IOException - { - if (isSslEnabled()) - { - return getSSLFactory().createSocket(arg1, arg2); - } - else - { - return new Socket(arg1, arg2); - } - } - - public Socket createSocket(InetAddress arg1, int arg2, InetAddress arg3, - int arg4) throws IOException - { - if (isSslEnabled()) - { - return getSSLFactory().createSocket(arg1, arg2, arg3, arg4); - } - else - { - return new Socket(arg1, arg2, arg3, arg4); - } - } - - public static javax.net.SocketFactory getSocketFactory() - { - if (factory == null) - { - factory = new SSLSocketFactory(); - } - return factory; - } - - private javax.net.ssl.SSLSocketFactory getSSLFactory() - { - if (sslFactory == null) - { - try - { - sslFactory = BogusSSLContextFactory.getInstance(false) - .getSocketFactory(); - } - catch (GeneralSecurityException e) - { - throw new RuntimeException("could not create SSL socket", e); - } - } - return sslFactory; - } - - public static boolean isSslEnabled() - { - return sslEnabled; - } - - public static void setSslEnabled(boolean newSslEnabled) - { - sslEnabled = newSslEnabled; - } - -} diff --git a/java/common/src/org/apache/qpid/url/AMQBindingURL.java b/java/common/src/org/apache/qpid/url/AMQBindingURL.java deleted file mode 100644 index b6a0bd500a..0000000000 --- a/java/common/src/org/apache/qpid/url/AMQBindingURL.java +++ /dev/null @@ -1,263 +0,0 @@ -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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.url; - -import org.apache.qpid.url.BindingURL; -import org.apache.qpid.url.URLHelper; -import org.apache.qpid.exchange.ExchangeDefaults; - -import java.util.HashMap; -import java.net.URI; -import java.net.URISyntaxException; - -public class AMQBindingURL implements BindingURL -{ - String _url; - String _exchangeClass; - String _exchangeName; - String _destinationName; - String _queueName; - private HashMap _options; - - - public AMQBindingURL(String url) throws URLSyntaxException - { - //format: - // :///[]/[]?