From 65b1a1ddfe95b9e273d4cdaf23067a0aaff9b1d1 Mon Sep 17 00:00:00 2001 From: Robert Godfrey Date: Fri, 7 Feb 2014 16:57:49 +0000 Subject: QPID-5504 : Refactoring to allow for nodes other than queues to be subscribed from, and nodes other than exchanges to be sent to (merged from separate branch) git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1565726 13f79535-47bb-0310-9956-ffa450edef68 --- .../qpid/server/protocol/v1_0/Connection_1_0.java | 25 +- .../server/protocol/v1_0/ConsumerTarget_1_0.java | 526 ++++++++++++++++ .../protocol/v1_0/MessageConverter_to_1_0.java | 3 +- .../server/protocol/v1_0/MessageMetaData_1_0.java | 4 +- .../protocol/v1_0/MessageSourceDestination.java | 59 ++ .../protocol/v1_0/NodeReceivingDestination.java | 106 ++++ .../server/protocol/v1_0/QueueDestination.java | 14 +- .../qpid/server/protocol/v1_0/SendingLink_1_0.java | 114 ++-- .../qpid/server/protocol/v1_0/Session_1_0.java | 25 +- .../server/protocol/v1_0/Subscription_1_0.java | 694 --------------------- 10 files changed, 796 insertions(+), 774 deletions(-) create mode 100644 qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java create mode 100644 qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageSourceDestination.java create mode 100644 qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NodeReceivingDestination.java delete mode 100644 qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Subscription_1_0.java (limited to 'qpid/java/broker-plugins/amqp-1-0-protocol') diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java index 4082f22e9c..41e2fef03f 100644 --- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java +++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java @@ -34,6 +34,7 @@ import org.apache.qpid.server.model.Transport; import org.apache.qpid.server.protocol.AMQConnectionModel; import org.apache.qpid.server.protocol.AMQSessionModel; import org.apache.qpid.server.stats.StatisticsCounter; +import org.apache.qpid.server.util.Action; import org.apache.qpid.server.virtualhost.VirtualHost; import java.util.ArrayList; @@ -53,16 +54,8 @@ public class Connection_1_0 implements ConnectionEventListener private final Collection _sessions = Collections.synchronizedCollection(new ArrayList()); private final Object _reference = new Object(); - - - public static interface Task - { - public void doTask(Connection_1_0 connection); - } - - - private List _closeTasks = - Collections.synchronizedList(new ArrayList()); + private List> _closeTasks = + Collections.synchronizedList(new ArrayList>()); @@ -98,26 +91,26 @@ public class Connection_1_0 implements ConnectionEventListener _sessions.remove(session); } - void removeConnectionCloseTask(final Task task) + void removeConnectionCloseTask(final Action task) { _closeTasks.remove( task ); } - void addConnectionCloseTask(final Task task) + void addConnectionCloseTask(final Action task) { _closeTasks.add( task ); } public void closeReceived() { - List taskCopy; + List> taskCopy; synchronized (_closeTasks) { - taskCopy = new ArrayList(_closeTasks); + taskCopy = new ArrayList>(_closeTasks); } - for(Task task : taskCopy) + for(Action task : taskCopy) { - task.doTask(this); + task.performAction(this); } synchronized (_closeTasks) { diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java new file mode 100644 index 0000000000..027c40aabe --- /dev/null +++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java @@ -0,0 +1,526 @@ +/* + * + * 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.server.protocol.v1_0; + +import org.apache.qpid.AMQException; +import org.apache.qpid.amqp_1_0.codec.ValueHandler; +import org.apache.qpid.amqp_1_0.messaging.SectionEncoder; +import org.apache.qpid.amqp_1_0.messaging.SectionEncoderImpl; +import org.apache.qpid.amqp_1_0.transport.SendingLinkEndpoint; +import org.apache.qpid.amqp_1_0.type.AmqpErrorException; +import org.apache.qpid.amqp_1_0.type.Binary; +import org.apache.qpid.amqp_1_0.type.DeliveryState; +import org.apache.qpid.amqp_1_0.type.Outcome; +import org.apache.qpid.amqp_1_0.type.UnsignedInteger; +import org.apache.qpid.amqp_1_0.type.codec.AMQPDescribedTypeRegistry; +import org.apache.qpid.amqp_1_0.type.messaging.Accepted; +import org.apache.qpid.amqp_1_0.type.messaging.Header; +import org.apache.qpid.amqp_1_0.type.messaging.Modified; +import org.apache.qpid.amqp_1_0.type.messaging.Released; +import org.apache.qpid.amqp_1_0.type.transaction.TransactionalState; +import org.apache.qpid.amqp_1_0.type.transport.SenderSettleMode; +import org.apache.qpid.amqp_1_0.type.transport.Transfer; +import org.apache.qpid.server.message.MessageInstance; +import org.apache.qpid.server.message.ServerMessage; +import org.apache.qpid.server.plugin.MessageConverter; +import org.apache.qpid.server.protocol.AMQSessionModel; +import org.apache.qpid.server.protocol.MessageConverterRegistry; +import org.apache.qpid.server.consumer.AbstractConsumerTarget; +import org.apache.qpid.server.consumer.Consumer; +import org.apache.qpid.server.txn.ServerTransaction; + +import java.nio.ByteBuffer; +import java.util.List; + +class ConsumerTarget_1_0 extends AbstractConsumerTarget +{ + private final boolean _acquires; + private SendingLink_1_0 _link; + + private long _deliveryTag = 0L; + + private Binary _transactionId; + private final AMQPDescribedTypeRegistry _typeRegistry; + private final SectionEncoder _sectionEncoder; + private Consumer _consumer; + + public ConsumerTarget_1_0(final SendingLink_1_0 link, + boolean acquires) + { + super(State.SUSPENDED); + _link = link; + _typeRegistry = link.getEndpoint().getSession().getConnection().getDescribedTypeRegistry(); + _sectionEncoder = new SectionEncoderImpl(_typeRegistry); + _acquires = acquires; + } + + public Consumer getConsumer() + { + return _consumer; + } + + private SendingLinkEndpoint getEndpoint() + { + return _link.getEndpoint(); + } + + public boolean isSuspended() + { + return _link.getSession().getConnectionModel().isStopped() || getState() != State.ACTIVE;// || !getEndpoint().hasCreditToSend(); + + } + + public boolean close() + { + boolean closed = false; + State state = getState(); + + getConsumer().getSendLock(); + try + { + while(!closed && state != State.CLOSED) + { + closed = updateState(state, State.CLOSED); + if(!closed) + { + state = getState(); + } + } + return closed; + } + finally + { + getConsumer().releaseSendLock(); + } + } + + public void send(MessageInstance entry, boolean batch) throws AMQException + { + // TODO + send(entry); + } + + public void flushBatched() + { + // TODO + } + + public void send(final MessageInstance queueEntry) throws AMQException + { + ServerMessage serverMessage = queueEntry.getMessage(); + Message_1_0 message; + if(serverMessage instanceof Message_1_0) + { + message = (Message_1_0) serverMessage; + } + else + { + final MessageConverter converter = MessageConverterRegistry.getConverter(serverMessage.getClass(), Message_1_0.class); + message = (Message_1_0) converter.convert(serverMessage, _link.getVirtualHost()); + } + + Transfer transfer = new Transfer(); + //TODO + + + List fragments = message.getFragments(); + ByteBuffer payload; + if(fragments.size() == 1) + { + payload = fragments.get(0); + } + else + { + int size = 0; + for(ByteBuffer fragment : fragments) + { + size += fragment.remaining(); + } + + payload = ByteBuffer.allocate(size); + + for(ByteBuffer fragment : fragments) + { + payload.put(fragment.duplicate()); + } + + payload.flip(); + } + + if(queueEntry.getDeliveryCount() != 0) + { + payload = payload.duplicate(); + ValueHandler valueHandler = new ValueHandler(_typeRegistry); + + Header oldHeader = null; + try + { + ByteBuffer encodedBuf = payload.duplicate(); + Object value = valueHandler.parse(payload); + if(value instanceof Header) + { + oldHeader = (Header) value; + } + else + { + payload.position(0); + } + } + catch (AmqpErrorException e) + { + //TODO + throw new RuntimeException(e); + } + + Header header = new Header(); + if(oldHeader != null) + { + header.setDurable(oldHeader.getDurable()); + header.setPriority(oldHeader.getPriority()); + header.setTtl(oldHeader.getTtl()); + } + header.setDeliveryCount(UnsignedInteger.valueOf(queueEntry.getDeliveryCount())); + _sectionEncoder.reset(); + _sectionEncoder.encodeObject(header); + Binary encodedHeader = _sectionEncoder.getEncoding(); + + ByteBuffer oldPayload = payload; + payload = ByteBuffer.allocate(oldPayload.remaining() + encodedHeader.getLength()); + payload.put(encodedHeader.getArray(),encodedHeader.getArrayOffset(),encodedHeader.getLength()); + payload.put(oldPayload); + payload.flip(); + } + + transfer.setPayload(payload); + byte[] data = new byte[8]; + ByteBuffer.wrap(data).putLong(_deliveryTag++); + final Binary tag = new Binary(data); + + transfer.setDeliveryTag(tag); + + synchronized(_link.getLock()) + { + if(_link.isAttached()) + { + if(SenderSettleMode.SETTLED.equals(getEndpoint().getSendingSettlementMode())) + { + transfer.setSettled(true); + } + else + { + UnsettledAction action = _acquires + ? new DispositionAction(tag, queueEntry) + : new DoNothingAction(tag, queueEntry); + + _link.addUnsettled(tag, action, queueEntry); + } + + if(_transactionId != null) + { + TransactionalState state = new TransactionalState(); + state.setTxnId(_transactionId); + transfer.setState(state); + } + // TODO - need to deal with failure here + if(_acquires && _transactionId != null) + { + ServerTransaction txn = _link.getTransaction(_transactionId); + if(txn != null) + { + txn.addPostTransactionAction(new ServerTransaction.Action(){ + + public void postCommit() + { + //To change body of implemented methods use File | Settings | File Templates. + } + + public void onRollback() + { + if(queueEntry.isAcquiredBy(getConsumer())) + { + queueEntry.release(); + _link.getEndpoint().updateDisposition(tag, (DeliveryState)null, true); + + + } + } + }); + } + + } + getSession().getConnectionModel().registerMessageDelivered(message.getSize()); + getEndpoint().transfer(transfer); + } + else + { + queueEntry.release(); + } + } + + } + + public void queueDeleted() + { + //TODO + getEndpoint().setSource(null); + getEndpoint().detach(); + } + + public boolean allocateCredit(final ServerMessage msg) + { + synchronized (_link.getLock()) + { + final boolean hasCredit = _link.isAttached() && getEndpoint().hasCreditToSend(); + if(!hasCredit && getState() == State.ACTIVE) + { + suspend(); + } + + return hasCredit; + } + } + + + public void suspend() + { + synchronized(_link.getLock()) + { + updateState(State.ACTIVE, State.SUSPENDED); + } + } + + + public void restoreCredit(final ServerMessage message) + { + //TODO + } + + public void queueEmpty() + { + synchronized(_link.getLock()) + { + if(_link.drained()) + { + updateState(State.ACTIVE, State.SUSPENDED); + } + } + } + + public void flowStateChanged() + { + synchronized(_link.getLock()) + { + if(isSuspended() && getEndpoint() != null) + { + updateState(State.SUSPENDED, State.ACTIVE); + _transactionId = _link.getTransactionId(); + } + } + } + + public Session_1_0 getSession() + { + return _link.getSession(); + } + + private class DispositionAction implements UnsettledAction + { + + private final MessageInstance _queueEntry; + private final Binary _deliveryTag; + + public DispositionAction(Binary tag, MessageInstance queueEntry) + { + _deliveryTag = tag; + _queueEntry = queueEntry; + } + + public boolean process(DeliveryState state, final Boolean settled) + { + + Binary transactionId = null; + final Outcome outcome; + // If disposition is settled this overrides the txn? + if(state instanceof TransactionalState) + { + transactionId = ((TransactionalState)state).getTxnId(); + outcome = ((TransactionalState)state).getOutcome(); + } + else if (state instanceof Outcome) + { + outcome = (Outcome) state; + } + else + { + outcome = null; + } + + + ServerTransaction txn = _link.getTransaction(transactionId); + + if(outcome instanceof Accepted) + { + txn.dequeue(_queueEntry.getOwningResource(), _queueEntry.getMessage(), + new ServerTransaction.Action() + { + + public void postCommit() + { + if(_queueEntry.isAcquiredBy(getConsumer())) + { + _queueEntry.delete(); + } + } + + public void onRollback() + { + + } + }); + txn.addPostTransactionAction(new ServerTransaction.Action() + { + public void postCommit() + { + //_link.getEndpoint().settle(_deliveryTag); + _link.getEndpoint().updateDisposition(_deliveryTag, (DeliveryState)outcome, true); + _link.getEndpoint().sendFlowConditional(); + } + + public void onRollback() + { + if(Boolean.TRUE.equals(settled)) + { + final Modified modified = new Modified(); + modified.setDeliveryFailed(true); + _link.getEndpoint().updateDisposition(_deliveryTag, modified, true); + _link.getEndpoint().sendFlowConditional(); + } + } + }); + } + else if(outcome instanceof Released) + { + txn.addPostTransactionAction(new ServerTransaction.Action() + { + public void postCommit() + { + + _queueEntry.release(); + _link.getEndpoint().settle(_deliveryTag); + } + + public void onRollback() + { + _link.getEndpoint().settle(_deliveryTag); + } + }); + } + + else if(outcome instanceof Modified) + { + txn.addPostTransactionAction(new ServerTransaction.Action() + { + public void postCommit() + { + + _queueEntry.release(); + if(Boolean.TRUE.equals(((Modified)outcome).getDeliveryFailed())) + { + _queueEntry.incrementDeliveryCount(); + } + _link.getEndpoint().settle(_deliveryTag); + } + + public void onRollback() + { + if(Boolean.TRUE.equals(settled)) + { + final Modified modified = new Modified(); + modified.setDeliveryFailed(true); + _link.getEndpoint().updateDisposition(_deliveryTag, modified, true); + _link.getEndpoint().sendFlowConditional(); + } + } + }); + } + + return (transactionId == null && outcome != null); + } + } + + private class DoNothingAction implements UnsettledAction + { + public DoNothingAction(final Binary tag, + final MessageInstance queueEntry) + { + } + + public boolean process(final DeliveryState state, final Boolean settled) + { + Binary transactionId = null; + Outcome outcome = null; + // If disposition is settled this overrides the txn? + if(state instanceof TransactionalState) + { + transactionId = ((TransactionalState)state).getTxnId(); + outcome = ((TransactionalState)state).getOutcome(); + } + else if (state instanceof Outcome) + { + outcome = (Outcome) state; + } + return true; + } + } + + @Override + public AMQSessionModel getSessionModel() + { + return getSession(); + } + + @Override + public void consumerAdded(final Consumer sub) + { + _consumer = sub; + } + + @Override + public void consumerRemoved(final Consumer sub) + { + + } + + @Override + public long getUnacknowledgedBytes() + { + // TODO + return 0; + } + + @Override + public long getUnacknowledgedMessages() + { + // TODO + return 0; + } + +} diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_to_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_to_1_0.java index 78ca9ff2a6..a96d951de6 100644 --- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_to_1_0.java +++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_to_1_0.java @@ -23,7 +23,6 @@ package org.apache.qpid.server.protocol.v1_0; import java.io.EOFException; import java.nio.ByteBuffer; import java.util.ArrayList; -import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; import java.util.ListIterator; @@ -286,7 +285,7 @@ public abstract class MessageConverter_to_1_0 implement Binary dataEncoding = sectionEncoder.getEncoding(); final ByteBuffer allData = ByteBuffer.allocate(headerSize + dataEncoding.getLength()); - metaData.writeToBuffer(0,allData); + metaData.writeToBuffer(allData); allData.put(dataEncoding.getArray(),dataEncoding.getArrayOffset(),dataEncoding.getLength()); return allData; } diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageMetaData_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageMetaData_1_0.java index 5026007360..be9d7a2d60 100755 --- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageMetaData_1_0.java +++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageMetaData_1_0.java @@ -314,7 +314,7 @@ public class MessageMetaData_1_0 implements StorableMessageMetaData return buf; } - public int writeToBuffer(int offsetInMetaData, ByteBuffer dest) + public int writeToBuffer(ByteBuffer dest) { ByteBuffer buf = _encoded; @@ -326,7 +326,7 @@ public class MessageMetaData_1_0 implements StorableMessageMetaData buf = buf.duplicate(); - buf.position(offsetInMetaData); + buf.position(0); if(dest.remaining() < buf.limit()) { diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageSourceDestination.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageSourceDestination.java new file mode 100644 index 0000000000..6f37d2d831 --- /dev/null +++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageSourceDestination.java @@ -0,0 +1,59 @@ +/* + * + * 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.server.protocol.v1_0; + +import org.apache.log4j.Logger; +import org.apache.qpid.amqp_1_0.type.Outcome; +import org.apache.qpid.amqp_1_0.type.messaging.Accepted; +import org.apache.qpid.server.message.MessageSource; +import org.apache.qpid.server.txn.ServerTransaction; + +public class MessageSourceDestination implements SendingDestination +{ + private static final Logger _logger = Logger.getLogger(MessageSourceDestination.class); + private static final Accepted ACCEPTED = new Accepted(); + private static final Outcome[] OUTCOMES = new Outcome[] { ACCEPTED }; + + + private MessageSource _queue; + + public MessageSourceDestination(MessageSource queue) + { + _queue = queue; + } + + public Outcome[] getOutcomes() + { + return OUTCOMES; + } + + public int getCredit() + { + // TODO - fix + return 100; + } + + public MessageSource getQueue() + { + return _queue; + } + +} diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NodeReceivingDestination.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NodeReceivingDestination.java new file mode 100644 index 0000000000..70f659b546 --- /dev/null +++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NodeReceivingDestination.java @@ -0,0 +1,106 @@ +/* + * + * 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.server.protocol.v1_0; + +import org.apache.qpid.amqp_1_0.type.Outcome; +import org.apache.qpid.amqp_1_0.type.messaging.Accepted; +import org.apache.qpid.amqp_1_0.type.messaging.Rejected; +import org.apache.qpid.amqp_1_0.type.messaging.TerminusDurability; +import org.apache.qpid.amqp_1_0.type.messaging.TerminusExpiryPolicy; +import org.apache.qpid.server.exchange.Exchange; +import org.apache.qpid.server.message.InstanceProperties; +import org.apache.qpid.server.message.MessageDestination; +import org.apache.qpid.server.txn.ServerTransaction; + +public class NodeReceivingDestination implements ReceivingDestination +{ + private static final Accepted ACCEPTED = new Accepted(); + public static final Rejected REJECTED = new Rejected(); + private static final Outcome[] OUTCOMES = { ACCEPTED, REJECTED}; + + private MessageDestination _exchange; + private TerminusDurability _durability; + private TerminusExpiryPolicy _expiryPolicy; + + public NodeReceivingDestination(MessageDestination exchange, TerminusDurability durable, TerminusExpiryPolicy expiryPolicy) + { + _exchange = exchange; + _durability = durable; + _expiryPolicy = expiryPolicy; + } + + public Outcome[] getOutcomes() + { + return OUTCOMES; + } + + public Outcome send(final Message_1_0 message, ServerTransaction txn) + { + final InstanceProperties instanceProperties = + new InstanceProperties() + { + + @Override + public Object getProperty(final Property prop) + { + switch(prop) + { + case MANDATORY: + return false; + case REDELIVERED: + return false; + case PERSISTENT: + return message.isPersistent(); + case IMMEDIATE: + return false; + case EXPIRATION: + return message.getExpiration(); + } + return null; + }}; + + int enqueues = _exchange.send(message, instanceProperties, txn, null); + + + return enqueues == 0 ? REJECTED : ACCEPTED; + } + + TerminusDurability getDurability() + { + return _durability; + } + + TerminusExpiryPolicy getExpiryPolicy() + { + return _expiryPolicy; + } + + public int getCredit() + { + // TODO - fix + return 20000; + } + + public MessageDestination getDestination() + { + return _exchange; + } +} diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java index b9c10b925f..3d6bb5e3db 100644 --- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java +++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java @@ -24,22 +24,21 @@ import org.apache.log4j.Logger; import org.apache.qpid.amqp_1_0.type.Outcome; import org.apache.qpid.amqp_1_0.type.messaging.Accepted; +import org.apache.qpid.server.message.MessageSource; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.txn.ServerTransaction; -public class QueueDestination implements SendingDestination, ReceivingDestination +public class QueueDestination extends MessageSourceDestination implements SendingDestination, ReceivingDestination { private static final Logger _logger = Logger.getLogger(QueueDestination.class); private static final Accepted ACCEPTED = new Accepted(); private static final Outcome[] OUTCOMES = new Outcome[] { ACCEPTED }; - private AMQQueue _queue; - public QueueDestination(AMQQueue queue) { - _queue = queue; + super(queue); } public Outcome[] getOutcomes() @@ -52,7 +51,7 @@ public class QueueDestination implements SendingDestination, ReceivingDestinatio try { - txn.enqueue(_queue,message, new ServerTransaction.Action() + txn.enqueue(getQueue(),message, new ServerTransaction.Action() { @@ -60,8 +59,7 @@ public class QueueDestination implements SendingDestination, ReceivingDestinatio { try { - - _queue.enqueue(message); + getQueue().enqueue(message,null); } catch (Exception e) { @@ -93,7 +91,7 @@ public class QueueDestination implements SendingDestination, ReceivingDestinatio public AMQQueue getQueue() { - return _queue; + return (AMQQueue) super.getQueue(); } } diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java index 4abf1bf76b..9e0327fe76 100644 --- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java +++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java @@ -22,6 +22,7 @@ package org.apache.qpid.server.protocol.v1_0; import java.util.ArrayList; import java.util.Collections; +import java.util.EnumSet; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -64,11 +65,14 @@ import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.exchange.TopicExchange; import org.apache.qpid.server.filter.JMSSelectorFilter; import org.apache.qpid.server.filter.SimpleFilterManager; +import org.apache.qpid.server.message.MessageInstance; +import org.apache.qpid.server.message.MessageSource; import org.apache.qpid.server.model.UUIDGenerator; import org.apache.qpid.server.queue.AMQQueue; -import org.apache.qpid.server.queue.QueueEntry; +import org.apache.qpid.server.consumer.Consumer; import org.apache.qpid.server.txn.AutoCommitTransaction; import org.apache.qpid.server.txn.ServerTransaction; +import org.apache.qpid.server.util.Action; import org.apache.qpid.server.virtualhost.VirtualHost; public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryStateHandler @@ -78,18 +82,22 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS private VirtualHost _vhost; private SendingDestination _destination; - private Subscription_1_0 _subscription; + private Consumer _consumer; + private ConsumerTarget_1_0 _target; + private boolean _draining; - private final Map _unsettledMap = - new HashMap(); + private final Map _unsettledMap = + new HashMap(); private final ConcurrentHashMap _unsettledActionMap = new ConcurrentHashMap(); private volatile SendingLinkAttachment _linkAttachment; private TerminusDurability _durability; - private List _resumeFullTransfers = new ArrayList(); + private List _resumeFullTransfers = new ArrayList(); private List _resumeAcceptedTransfers = new ArrayList(); private Runnable _closeAction; + private final MessageSource _queue; + public SendingLink_1_0(final SendingLinkAttachment linkAttachment, final VirtualHost vhost, @@ -103,24 +111,22 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS _durability = source.getDurable(); linkAttachment.setDeliveryStateHandler(this); QueueDestination qd = null; - AMQQueue queue = null; + EnumSet options = EnumSet.noneOf(Consumer.Option.class); boolean noLocal = false; JMSSelectorFilter messageFilter = null; - if(destination instanceof QueueDestination) + if(destination instanceof MessageSourceDestination) { - queue = ((QueueDestination) _destination).getQueue(); + _queue = ((MessageSourceDestination) _destination).getQueue(); - if(queue.getAvailableAttributes().contains("topic")) + if(_queue instanceof AMQQueue && ((AMQQueue)_queue).getAvailableAttributes().contains("topic")) { source.setDistributionMode(StdDistMode.COPY); } - qd = (QueueDestination) destination; - Map filters = source.getFilter(); Map actualFilters = new HashMap(); @@ -167,7 +173,13 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS } source.setFilter(actualFilters.isEmpty() ? null : actualFilters); - _subscription = new Subscription_1_0(this, qd, source.getDistributionMode() != StdDistMode.COPY); + _target = new ConsumerTarget_1_0(this, source.getDistributionMode() != StdDistMode.COPY); + if(source.getDistributionMode() != StdDistMode.COPY) + { + options.add(Consumer.Option.ACQUIRES); + options.add(Consumer.Option.SEES_REQUEUES); + } + } else if(destination instanceof ExchangeDestination) { @@ -199,7 +211,7 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS name = UUID.randomUUID().toString(); } - queue = _vhost.getQueue(name); + AMQQueue queue = _vhost.getQueue(name); Exchange exchange = exchangeDestination.getExchange(); if(queue == null) @@ -299,9 +311,10 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS } } } + _queue = queue; source.setFilter(actualFilters.isEmpty() ? null : actualFilters); - exchange.addBinding(binding,queue,null); + exchange.addBinding(binding, queue,null); source.setDistributionMode(StdDistMode.COPY); if(!isDurable) @@ -309,10 +322,10 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS final String queueName = name; final AMQQueue tempQueue = queue; - final Connection_1_0.Task deleteQueueTask = - new Connection_1_0.Task() + final Action deleteQueueTask = + new Action() { - public void doTask(Connection_1_0 session) + public void performAction(Connection_1_0 session) { if (_vhost.getQueue(queueName) == tempQueue) { @@ -331,9 +344,9 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS getSession().getConnection().addConnectionCloseTask(deleteQueueTask); - queue.addQueueDeleteTask(new AMQQueue.Task() + queue.addQueueDeleteTask(new Action() { - public void doTask(AMQQueue queue) + public void performAction(AMQQueue queue) { getSession().getConnection().removeConnectionCloseTask(deleteQueueTask); } @@ -347,31 +360,46 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS catch (AMQSecurityException e) { _logger.error("Security error", e); + throw new RuntimeException(e); } catch (AMQInternalException e) { _logger.error("Internal error", e); + throw new RuntimeException(e); } catch (AMQException e) { _logger.error("Error", e); + throw new RuntimeException(e); } - _subscription = new Subscription_1_0(this, qd, true); + + _target = new ConsumerTarget_1_0(this, true); + options.add(Consumer.Option.ACQUIRES); + options.add(Consumer.Option.SEES_REQUEUES); + + } + else + { + throw new RuntimeException("Unknown destination type"); } - if(_subscription != null) + if(_target != null) { - _subscription.setNoLocal(noLocal); - if(messageFilter!=null) + if(noLocal) { - _subscription.setFilters(new SimpleFilterManager(messageFilter)); + options.add(Consumer.Option.NO_LOCAL); } + + _consumer.setNoLocal(noLocal); + + try { - - queue.registerSubscription(_subscription, false); + _consumer = _queue.addConsumer(_target, + messageFilter == null ? null : new SimpleFilterManager(messageFilter), + Message_1_0.class, getEndpoint().getName(), options); } catch (AMQException e) { @@ -394,12 +422,11 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS // if not durable or close if(!TerminusDurability.UNSETTLED_STATE.equals(_durability)) { - AMQQueue queue = _subscription.getQueue(); try { - queue.unregisterSubscription(_subscription); + _consumer.close(); } catch (AMQException e) @@ -426,7 +453,7 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS { try { - queue.getVirtualHost().removeQueue(queue); + _vhost.removeQueue((AMQQueue)_queue); } catch(AMQException e) { @@ -443,7 +470,7 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS else if(detach == null || detach.getError() != null) { _linkAttachment = null; - _subscription.flowStateChanged(); + _target.flowStateChanged(); } else { @@ -491,7 +518,7 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS } if(_resumeAcceptedTransfers.isEmpty()) { - _subscription.flowStateChanged(); + _target.flowStateChanged(); } } @@ -531,7 +558,7 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS } } - public void addUnsettled(Binary tag, UnsettledAction unsettledAction, QueueEntry queueEntry) + public void addUnsettled(Binary tag, UnsettledAction unsettledAction, MessageInstance queueEntry) { _unsettledActionMap.put(tag,unsettledAction); if(getTransactionId() == null) @@ -593,9 +620,9 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS public synchronized void setLinkAttachment(SendingLinkAttachment linkAttachment) { - if(_subscription.isActive()) + if(_consumer.isActive()) { - _subscription.suspend(); + _target.suspend(); } _linkAttachment = linkAttachment; @@ -603,14 +630,14 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS SendingLinkEndpoint endpoint = linkAttachment.getEndpoint(); endpoint.setDeliveryStateHandler(this); Map initialUnsettledMap = endpoint.getInitialUnsettledMap(); - Map unsettledCopy = new HashMap(_unsettledMap); + Map unsettledCopy = new HashMap(_unsettledMap); _resumeAcceptedTransfers.clear(); _resumeFullTransfers.clear(); - for(Map.Entry entry : unsettledCopy.entrySet()) + for(Map.Entry entry : unsettledCopy.entrySet()) { Binary deliveryTag = entry.getKey(); - final QueueEntry queueEntry = entry.getValue(); + final MessageInstance queueEntry = entry.getValue(); if(initialUnsettledMap == null || !initialUnsettledMap.containsKey(deliveryTag)) { queueEntry.setRedelivered(); @@ -624,7 +651,7 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS if(outcome instanceof Accepted) { AutoCommitTransaction txn = new AutoCommitTransaction(_vhost.getMessageStore()); - if(_subscription.acquires()) + if(_consumer.acquires()) { txn.dequeue(Collections.singleton(queueEntry), new ServerTransaction.Action() @@ -644,7 +671,7 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS else if(outcome instanceof Released) { AutoCommitTransaction txn = new AutoCommitTransaction(_vhost.getMessageStore()); - if(_subscription.acquires()) + if(_consumer.acquires()) { txn.dequeue(Collections.singleton(queueEntry), new ServerTransaction.Action() @@ -678,9 +705,9 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS public Map getUnsettledOutcomeMap() { - Map unsettled = new HashMap(_unsettledMap); + Map unsettled = new HashMap(_unsettledMap); - for(Map.Entry entry : unsettled.entrySet()) + for(Map.Entry entry : unsettled.entrySet()) { entry.setValue(null); } @@ -692,4 +719,9 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS { _closeAction = action; } + + public VirtualHost getVirtualHost() + { + return _vhost; + } } diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java index 823e4cb16d..beed6be84b 100644 --- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java +++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java @@ -41,6 +41,8 @@ import org.apache.qpid.AMQSecurityException; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.logging.LogSubject; +import org.apache.qpid.server.message.MessageDestination; +import org.apache.qpid.server.message.MessageSource; import org.apache.qpid.server.model.UUIDGenerator; import org.apache.qpid.server.protocol.AMQConnectionModel; import org.apache.qpid.server.protocol.AMQSessionModel; @@ -48,6 +50,7 @@ import org.apache.qpid.server.protocol.LinkRegistry; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.txn.AutoCommitTransaction; import org.apache.qpid.server.txn.ServerTransaction; +import org.apache.qpid.server.util.Action; import org.apache.qpid.server.virtualhost.VirtualHost; import java.util.*; @@ -108,11 +111,11 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel, LogSu source.setAddress(tempQueue.getName()); } String addr = source.getAddress(); - AMQQueue queue = _vhost.getQueue(addr); + MessageSource queue = _vhost.getMessageSource(addr); if(queue != null) { - destination = new QueueDestination(queue); + destination = new MessageSourceDestination(queue); @@ -249,11 +252,11 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel, LogSu } String addr = target.getAddress(); - Exchange exchg = _vhost.getExchange(addr); - if(exchg != null) + MessageDestination messageDestination = _vhost.getMessageDestination(addr); + if(messageDestination != null) { - destination = new ExchangeDestination(exchg, target.getDurable(), - target.getExpiryPolicy()); + destination = new NodeReceivingDestination(messageDestination, target.getDurable(), + target.getExpiryPolicy()); } else { @@ -343,10 +346,10 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel, LogSu if (lifetimePolicy == null || lifetimePolicy instanceof DeleteOnClose) { - final Connection_1_0.Task deleteQueueTask = - new Connection_1_0.Task() + final Action deleteQueueTask = + new Action() { - public void doTask(Connection_1_0 session) + public void performAction(Connection_1_0 session) { if (_vhost.getQueue(queueName) == tempQueue) { @@ -365,9 +368,9 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel, LogSu _connection.addConnectionCloseTask(deleteQueueTask); - queue.addQueueDeleteTask(new AMQQueue.Task() + queue.addQueueDeleteTask(new Action() { - public void doTask(AMQQueue queue) + public void performAction(AMQQueue queue) { _connection.removeConnectionCloseTask(deleteQueueTask); } diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Subscription_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Subscription_1_0.java deleted file mode 100644 index 6a3f5b46e1..0000000000 --- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Subscription_1_0.java +++ /dev/null @@ -1,694 +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.server.protocol.v1_0; - -import java.nio.ByteBuffer; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicReference; -import java.util.concurrent.locks.ReentrantLock; -import org.apache.qpid.AMQException; -import org.apache.qpid.amqp_1_0.codec.ValueHandler; -import org.apache.qpid.amqp_1_0.messaging.SectionEncoder; -import org.apache.qpid.amqp_1_0.messaging.SectionEncoderImpl; -import org.apache.qpid.amqp_1_0.transport.SendingLinkEndpoint; -import org.apache.qpid.amqp_1_0.type.AmqpErrorException; -import org.apache.qpid.amqp_1_0.type.Binary; -import org.apache.qpid.amqp_1_0.type.DeliveryState; -import org.apache.qpid.amqp_1_0.type.Outcome; -import org.apache.qpid.amqp_1_0.type.UnsignedInteger; -import org.apache.qpid.amqp_1_0.type.codec.AMQPDescribedTypeRegistry; -import org.apache.qpid.amqp_1_0.type.messaging.Accepted; -import org.apache.qpid.amqp_1_0.type.messaging.Header; -import org.apache.qpid.amqp_1_0.type.messaging.Modified; -import org.apache.qpid.amqp_1_0.type.messaging.Released; -import org.apache.qpid.amqp_1_0.type.messaging.Source; -import org.apache.qpid.amqp_1_0.type.messaging.StdDistMode; -import org.apache.qpid.amqp_1_0.type.transaction.TransactionalState; -import org.apache.qpid.amqp_1_0.type.transport.SenderSettleMode; -import org.apache.qpid.amqp_1_0.type.transport.Transfer; -import org.apache.qpid.server.plugin.MessageConverter; -import org.apache.qpid.server.protocol.MessageConverterRegistry; -import org.apache.qpid.server.filter.FilterManager; -import org.apache.qpid.server.logging.LogActor; -import org.apache.qpid.server.message.ServerMessage; -import org.apache.qpid.server.protocol.AMQSessionModel; -import org.apache.qpid.server.queue.AMQQueue; -import org.apache.qpid.server.queue.QueueEntry; -import org.apache.qpid.server.subscription.Subscription; -import org.apache.qpid.server.txn.ServerTransaction; - -class - Subscription_1_0 implements Subscription -{ - private SendingLink_1_0 _link; - - private AMQQueue _queue; - - private final AtomicReference _state = new AtomicReference(State.SUSPENDED); - - private final QueueEntry.SubscriptionAcquiredState _owningState = new QueueEntry.SubscriptionAcquiredState(this); - private final long _id; - private final boolean _acquires; - private volatile AMQQueue.Context _queueContext; - private Map _properties = new ConcurrentHashMap(); - private ReentrantLock _stateChangeLock = new ReentrantLock(); - - private boolean _noLocal; - private FilterManager _filters; - - private long _deliveryTag = 0L; - private StateListener _stateListener; - - private Binary _transactionId; - private final AMQPDescribedTypeRegistry _typeRegistry = AMQPDescribedTypeRegistry.newInstance() - .registerTransportLayer() - .registerMessagingLayer() - .registerTransactionLayer() - .registerSecurityLayer(); - private SectionEncoder _sectionEncoder = new SectionEncoderImpl(_typeRegistry); - - public Subscription_1_0(final SendingLink_1_0 link, final QueueDestination destination) - { - this(link, destination, ((Source)link.getEndpoint().getSource()).getDistributionMode() != StdDistMode.COPY); - } - - public Subscription_1_0(final SendingLink_1_0 link, final QueueDestination destination, boolean acquires) - { - _link = link; - _queue = destination.getQueue(); - _id = getEndpoint().getLocalHandle().longValue(); - _acquires = acquires; - } - - private SendingLinkEndpoint getEndpoint() - { - return _link.getEndpoint(); - } - - public LogActor getLogActor() - { - return null; //TODO - } - - public boolean isTransient() - { - return true; //TODO - } - - public AMQQueue getQueue() - { - return _queue; - } - - public QueueEntry.SubscriptionAcquiredState getOwningState() - { - return _owningState; - } - - public void setQueue(final AMQQueue queue, final boolean exclusive) - { - //TODO - } - - public void setNoLocal(final boolean noLocal) - { - _noLocal = noLocal; - } - - public long getSubscriptionID() - { - return _id; - } - - public boolean isSuspended() - { - return _link.getSession().getConnectionModel().isStopped() || !isActive();// || !getEndpoint().hasCreditToSend(); - - } - - public boolean hasInterest(final QueueEntry entry) - { - if(_noLocal && entry.getMessage().getConnectionReference() == getSession().getConnection().getReference()) - { - return false; - } - else if(!(entry.getMessage() instanceof Message_1_0) - && MessageConverterRegistry.getConverter(entry.getMessage().getClass(), Message_1_0.class)==null) - { - return false; - } - return checkFilters(entry); - - } - - private boolean checkFilters(final QueueEntry entry) - { - return (_filters == null) || _filters.allAllow(entry.asFilterable()); - } - - public boolean isClosed() - { - return !getEndpoint().isAttached(); - } - - public boolean acquires() - { - return _acquires; - } - - public boolean seesRequeues() - { - // TODO - return acquires(); - } - - public void close() - { - getEndpoint().detach(); - } - - public void send(QueueEntry entry, boolean batch) throws AMQException - { - // TODO - send(entry); - } - - public void flushBatched() - { - // TODO - } - - public void send(final QueueEntry queueEntry) throws AMQException - { - ServerMessage serverMessage = queueEntry.getMessage(); - Message_1_0 message; - if(serverMessage instanceof Message_1_0) - { - message = (Message_1_0) serverMessage; - } - else - { - final MessageConverter converter = MessageConverterRegistry.getConverter(serverMessage.getClass(), Message_1_0.class); - message = (Message_1_0) converter.convert(serverMessage, queueEntry.getQueue().getVirtualHost()); - } - - Transfer transfer = new Transfer(); - //TODO - - - List fragments = message.getFragments(); - ByteBuffer payload; - if(fragments.size() == 1) - { - payload = fragments.get(0); - } - else - { - int size = 0; - for(ByteBuffer fragment : fragments) - { - size += fragment.remaining(); - } - - payload = ByteBuffer.allocate(size); - - for(ByteBuffer fragment : fragments) - { - payload.put(fragment.duplicate()); - } - - payload.flip(); - } - - if(queueEntry.getDeliveryCount() != 0) - { - payload = payload.duplicate(); - ValueHandler valueHandler = new ValueHandler(_typeRegistry); - - Header oldHeader = null; - try - { - ByteBuffer encodedBuf = payload.duplicate(); - Object value = valueHandler.parse(payload); - if(value instanceof Header) - { - oldHeader = (Header) value; - } - else - { - payload.position(0); - } - } - catch (AmqpErrorException e) - { - //TODO - throw new RuntimeException(e); - } - - Header header = new Header(); - if(oldHeader != null) - { - header.setDurable(oldHeader.getDurable()); - header.setPriority(oldHeader.getPriority()); - header.setTtl(oldHeader.getTtl()); - } - header.setDeliveryCount(UnsignedInteger.valueOf(queueEntry.getDeliveryCount())); - _sectionEncoder.reset(); - _sectionEncoder.encodeObject(header); - Binary encodedHeader = _sectionEncoder.getEncoding(); - - ByteBuffer oldPayload = payload; - payload = ByteBuffer.allocate(oldPayload.remaining() + encodedHeader.getLength()); - payload.put(encodedHeader.getArray(),encodedHeader.getArrayOffset(),encodedHeader.getLength()); - payload.put(oldPayload); - payload.flip(); - } - - transfer.setPayload(payload); - byte[] data = new byte[8]; - ByteBuffer.wrap(data).putLong(_deliveryTag++); - final Binary tag = new Binary(data); - - transfer.setDeliveryTag(tag); - - synchronized(_link.getLock()) - { - if(_link.isAttached()) - { - if(SenderSettleMode.SETTLED.equals(getEndpoint().getSendingSettlementMode())) - { - transfer.setSettled(true); - } - else - { - UnsettledAction action = _acquires - ? new DispositionAction(tag, queueEntry) - : new DoNothingAction(tag, queueEntry); - - _link.addUnsettled(tag, action, queueEntry); - } - - if(_transactionId != null) - { - TransactionalState state = new TransactionalState(); - state.setTxnId(_transactionId); - transfer.setState(state); - } - // TODO - need to deal with failure here - if(_acquires && _transactionId != null) - { - ServerTransaction txn = _link.getTransaction(_transactionId); - if(txn != null) - { - txn.addPostTransactionAction(new ServerTransaction.Action(){ - - public void postCommit() - { - //To change body of implemented methods use File | Settings | File Templates. - } - - public void onRollback() - { - if(queueEntry.isAcquiredBy(Subscription_1_0.this)) - { - queueEntry.release(); - _link.getEndpoint().updateDisposition(tag, (DeliveryState)null, true); - - - } - } - }); - } - - } - getSession().getConnectionModel().registerMessageDelivered(message.getSize()); - getEndpoint().transfer(transfer); - } - else - { - queueEntry.release(); - } - } - - } - - public void queueDeleted(final AMQQueue queue) - { - //TODO - getEndpoint().setSource(null); - getEndpoint().detach(); - } - - public boolean wouldSuspend(final QueueEntry msg) - { - synchronized (_link.getLock()) - { - final boolean hasCredit = _link.isAttached() && getEndpoint().hasCreditToSend(); - if(!hasCredit && getState() == State.ACTIVE) - { - suspend(); - } - - return !hasCredit; - } - } - - public boolean trySendLock() - { - return _stateChangeLock.tryLock(); - } - - public void suspend() - { - synchronized(_link.getLock()) - { - if(_state.compareAndSet(State.ACTIVE, State.SUSPENDED)) - { - _stateListener.stateChange(this, State.ACTIVE, State.SUSPENDED); - } - } - } - - public void getSendLock() - { - _stateChangeLock.lock(); - } - - public void releaseSendLock() - { - _stateChangeLock.unlock(); - } - - public void releaseQueueEntry(QueueEntry queueEntryImpl) - { - //To change body of implemented methods use File | Settings | File Templates. - } - - - public void onDequeue(final QueueEntry queueEntry) - { - //TODO - } - - public void restoreCredit(final QueueEntry queueEntry) - { - //TODO - } - - public void setStateListener(final StateListener listener) - { - _stateListener = listener; - } - - public State getState() - { - return _state.get(); - } - - public AMQQueue.Context getQueueContext() - { - return _queueContext; - } - - public void setQueueContext(AMQQueue.Context queueContext) - { - _queueContext = queueContext; - } - - - public boolean isActive() - { - return getState() == State.ACTIVE; - } - - public void set(String key, Object value) - { - _properties.put(key, value); - } - - public Object get(String key) - { - return _properties.get(key); - } - - public boolean isSessionTransactional() - { - return false; //TODO - } - - public void queueEmpty() - { - synchronized(_link.getLock()) - { - if(_link.drained()) - { - if(_state.compareAndSet(State.ACTIVE, State.SUSPENDED)) - { - _stateListener.stateChange(this, State.ACTIVE, State.SUSPENDED); - } - } - } - } - - public void flowStateChanged() - { - synchronized(_link.getLock()) - { - if(isSuspended() && getEndpoint() != null) - { - if(_state.compareAndSet(State.SUSPENDED, State.ACTIVE)) - { - _stateListener.stateChange(this, State.SUSPENDED, State.ACTIVE); - } - _transactionId = _link.getTransactionId(); - } - } - } - - public Session_1_0 getSession() - { - return _link.getSession(); - } - - private class DispositionAction implements UnsettledAction - { - - private final QueueEntry _queueEntry; - private final Binary _deliveryTag; - - public DispositionAction(Binary tag, QueueEntry queueEntry) - { - _deliveryTag = tag; - _queueEntry = queueEntry; - } - - public boolean process(DeliveryState state, final Boolean settled) - { - - Binary transactionId = null; - final Outcome outcome; - // If disposition is settled this overrides the txn? - if(state instanceof TransactionalState) - { - transactionId = ((TransactionalState)state).getTxnId(); - outcome = ((TransactionalState)state).getOutcome(); - } - else if (state instanceof Outcome) - { - outcome = (Outcome) state; - } - else - { - outcome = null; - } - - - ServerTransaction txn = _link.getTransaction(transactionId); - - if(outcome instanceof Accepted) - { - txn.dequeue(_queueEntry.getQueue(), _queueEntry.getMessage(), - new ServerTransaction.Action() - { - - public void postCommit() - { - if(_queueEntry.isAcquiredBy(Subscription_1_0.this)) - { - _queueEntry.delete(); - } - } - - public void onRollback() - { - - } - }); - txn.addPostTransactionAction(new ServerTransaction.Action() - { - public void postCommit() - { - //_link.getEndpoint().settle(_deliveryTag); - _link.getEndpoint().updateDisposition(_deliveryTag, (DeliveryState)outcome, true); - _link.getEndpoint().sendFlowConditional(); - } - - public void onRollback() - { - if(Boolean.TRUE.equals(settled)) - { - final Modified modified = new Modified(); - modified.setDeliveryFailed(true); - _link.getEndpoint().updateDisposition(_deliveryTag, modified, true); - _link.getEndpoint().sendFlowConditional(); - } - } - }); - } - else if(outcome instanceof Released) - { - txn.addPostTransactionAction(new ServerTransaction.Action() - { - public void postCommit() - { - - _queueEntry.release(); - _link.getEndpoint().settle(_deliveryTag); - } - - public void onRollback() - { - _link.getEndpoint().settle(_deliveryTag); - } - }); - } - - else if(outcome instanceof Modified) - { - txn.addPostTransactionAction(new ServerTransaction.Action() - { - public void postCommit() - { - - _queueEntry.release(); - if(Boolean.TRUE.equals(((Modified)outcome).getDeliveryFailed())) - { - _queueEntry.incrementDeliveryCount(); - } - _link.getEndpoint().settle(_deliveryTag); - } - - public void onRollback() - { - if(Boolean.TRUE.equals(settled)) - { - final Modified modified = new Modified(); - modified.setDeliveryFailed(true); - _link.getEndpoint().updateDisposition(_deliveryTag, modified, true); - _link.getEndpoint().sendFlowConditional(); - } - } - }); - } - - return (transactionId == null && outcome != null); - } - } - - private class DoNothingAction implements UnsettledAction - { - public DoNothingAction(final Binary tag, - final QueueEntry queueEntry) - { - } - - public boolean process(final DeliveryState state, final Boolean settled) - { - Binary transactionId = null; - Outcome outcome = null; - // If disposition is settled this overrides the txn? - if(state instanceof TransactionalState) - { - transactionId = ((TransactionalState)state).getTxnId(); - outcome = ((TransactionalState)state).getOutcome(); - } - else if (state instanceof Outcome) - { - outcome = (Outcome) state; - } - return true; - } - } - - public FilterManager getFilters() - { - return _filters; - } - - public void setFilters(final FilterManager filters) - { - _filters = filters; - } - - @Override - public AMQSessionModel getSessionModel() - { - // TODO - return getSession(); - } - - @Override - public long getBytesOut() - { - // TODO - return 0; - } - - @Override - public long getMessagesOut() - { - // TODO - return 0; - } - - @Override - public long getUnacknowledgedBytes() - { - // TODO - return 0; - } - - @Override - public long getUnacknowledgedMessages() - { - // TODO - return 0; - } - - @Override - public String getConsumerName() - { - //TODO - return "TODO"; - } -} -- cgit v1.2.1