001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.activemq.transport.amqp.protocol;
018
019import static org.apache.activemq.transport.amqp.AmqpSupport.toBytes;
020import static org.apache.activemq.transport.amqp.AmqpSupport.toLong;
021
022import java.io.IOException;
023import java.util.HashSet;
024import java.util.Set;
025
026import org.apache.activemq.command.ActiveMQDestination;
027import org.apache.activemq.command.ConnectionId;
028import org.apache.activemq.command.ExceptionResponse;
029import org.apache.activemq.command.LocalTransactionId;
030import org.apache.activemq.command.Response;
031import org.apache.activemq.command.TransactionInfo;
032import org.apache.activemq.transport.amqp.AmqpProtocolConverter;
033import org.apache.qpid.proton.Proton;
034import org.apache.qpid.proton.amqp.Binary;
035import org.apache.qpid.proton.amqp.Symbol;
036import org.apache.qpid.proton.amqp.messaging.Accepted;
037import org.apache.qpid.proton.amqp.messaging.AmqpValue;
038import org.apache.qpid.proton.amqp.messaging.Rejected;
039import org.apache.qpid.proton.amqp.transaction.Declare;
040import org.apache.qpid.proton.amqp.transaction.Declared;
041import org.apache.qpid.proton.amqp.transaction.Discharge;
042import org.apache.qpid.proton.amqp.transport.ErrorCondition;
043import org.apache.qpid.proton.engine.Delivery;
044import org.apache.qpid.proton.engine.Receiver;
045import org.apache.qpid.proton.message.Message;
046import org.fusesource.hawtbuf.Buffer;
047import org.slf4j.Logger;
048import org.slf4j.LoggerFactory;
049
050/**
051 * Implements the AMQP Transaction Coordinator support to manage local
052 * transactions between an AMQP client and the broker.
053 */
054public class AmqpTransactionCoordinator extends AmqpAbstractReceiver {
055
056    private static final Logger LOG = LoggerFactory.getLogger(AmqpTransactionCoordinator.class);
057
058    private final Set<AmqpSession> txSessions = new HashSet<AmqpSession>();
059
060    /**
061     * Creates a new Transaction coordinator used to manage AMQP transactions.
062     *
063     * @param session
064     *        the AmqpSession under which the coordinator was created.
065     * @param endpoint
066     *        the AMQP receiver link endpoint for this coordinator.
067     */
068    public AmqpTransactionCoordinator(AmqpSession session, Receiver endpoint) {
069        super(session, endpoint);
070    }
071
072    @Override
073    protected void processDelivery(final Delivery delivery, Buffer deliveryBytes) throws Exception {
074        Message message = Proton.message();
075        int offset = deliveryBytes.offset;
076        int len = deliveryBytes.length;
077
078        while (len > 0) {
079            final int decoded = message.decode(deliveryBytes.data, offset, len);
080            assert decoded > 0 : "Make progress decoding the message";
081            offset += decoded;
082            len -= decoded;
083        }
084
085        final AmqpSession session = (AmqpSession) getEndpoint().getSession().getContext();
086        final ConnectionId connectionId = session.getConnection().getConnectionId();
087        final Object action = ((AmqpValue) message.getBody()).getValue();
088
089        LOG.debug("COORDINATOR received: {}, [{}]", action, deliveryBytes);
090        if (action instanceof Declare) {
091            Declare declare = (Declare) action;
092            if (declare.getGlobalId() != null) {
093                throw new Exception("don't know how to handle a declare /w a set GlobalId");
094            }
095
096            LocalTransactionId txId = session.getConnection().getNextTransactionId();
097            TransactionInfo txInfo = new TransactionInfo(connectionId, txId, TransactionInfo.BEGIN);
098            session.getConnection().registerTransaction(txId, this);
099            sendToActiveMQ(txInfo, null);
100            LOG.trace("started transaction {}", txId);
101
102            Declared declared = new Declared();
103            declared.setTxnId(new Binary(toBytes(txId.getValue())));
104            delivery.disposition(declared);
105            delivery.settle();
106        } else if (action instanceof Discharge) {
107            final Discharge discharge = (Discharge) action;
108            final LocalTransactionId txId = new LocalTransactionId(connectionId, toLong(discharge.getTxnId()));
109            final byte operation;
110
111            if (discharge.getFail()) {
112                LOG.trace("rollback transaction {}", txId);
113                operation = TransactionInfo.ROLLBACK;
114            } else {
115                LOG.trace("commit transaction {}", txId);
116                operation = TransactionInfo.COMMIT_ONE_PHASE;
117            }
118
119            for (AmqpSession txSession : txSessions) {
120                if (operation == TransactionInfo.ROLLBACK) {
121                    txSession.rollback(txId);
122                } else {
123                    txSession.commit(txId);
124                }
125            }
126
127            txSessions.clear();
128            session.getConnection().unregisterTransaction(txId);
129
130            TransactionInfo txinfo = new TransactionInfo(connectionId, txId, operation);
131            sendToActiveMQ(txinfo, new ResponseHandler() {
132                @Override
133                public void onResponse(AmqpProtocolConverter converter, Response response) throws IOException {
134                    if (response.isException()) {
135                        ExceptionResponse er = (ExceptionResponse) response;
136                        Rejected rejected = new Rejected();
137                        rejected.setError(new ErrorCondition(Symbol.valueOf("failed"), er.getException().getMessage()));
138                        delivery.disposition(rejected);
139                    } else {
140                        delivery.disposition(Accepted.getInstance());
141                    }
142
143                    LOG.debug("TX: {} settling {}", operation, action);
144                    delivery.settle();
145                    session.pumpProtonToSocket();
146                }
147            });
148
149            if (operation == TransactionInfo.ROLLBACK) {
150                session.flushPendingMessages();
151            }
152
153        } else {
154            throw new Exception("Expected coordinator message type: " + action.getClass());
155        }
156
157        replenishCredit();
158    }
159
160    private void replenishCredit() {
161        if (getEndpoint().getCredit() <= (getConfiguredReceiverCredit() * .2)) {
162            LOG.debug("Sending more credit ({}) to transaction coordinator on session {}", getConfiguredReceiverCredit() - getEndpoint().getCredit(), session.getSessionId());
163            getEndpoint().flow(getConfiguredReceiverCredit() - getEndpoint().getCredit());
164            session.pumpProtonToSocket();
165        }
166    }
167
168    @Override
169    public ActiveMQDestination getDestination() {
170        return null;
171    }
172
173    @Override
174    public void setDestination(ActiveMQDestination destination) {
175    }
176
177    public void enlist(AmqpSession session) {
178        txSessions.add(session);
179    }
180}