path: root/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/SubscriptionTarget_0_8.java

/*
 *
 * 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.v0_8;

import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
import org.apache.qpid.common.AMQPFilterTypes;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.server.flow.FlowCreditManager;
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.messages.SubscriptionMessages;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.v0_8.output.ProtocolOutputConverter;
import org.apache.qpid.server.queue.QueueEntry;
import org.apache.qpid.server.subscription.AbstractSubscriptionTarget;
import org.apache.qpid.server.subscription.ClientDeliveryMethod;
import org.apache.qpid.server.subscription.RecordDeliveryMethod;
import org.apache.qpid.server.subscription.Subscription;
import org.apache.qpid.server.txn.AutoCommitTransaction;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.StateChangeListener;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Encapsulation of a subscription to a queue. <p/> Ties together the protocol session of a subscriber, the consumer tag
 * that was given out by the broker and the channel id. <p/>
 */
public abstract class SubscriptionTarget_0_8 extends AbstractSubscriptionTarget implements FlowCreditManager.FlowCreditManagerListener
{

    private final StateChangeListener<QueueEntry, QueueEntry.State> _entryReleaseListener =
            new StateChangeListener<QueueEntry, QueueEntry.State>()
            {
                @Override
                public void stateChanged(final QueueEntry entry,
                                         final QueueEntry.State oldSate,
                                         final QueueEntry.State newState)
                {
                    if (oldSate == QueueEntry.State.ACQUIRED && (newState == QueueEntry.State.AVAILABLE || newState == QueueEntry.State.DEQUEUED))
                    {
                        restoreCredit(entry);
                    }
                    entry.removeStateChangeListener(this);
                }
            };

    private final ClientDeliveryMethod _deliveryMethod;
    private final RecordDeliveryMethod _recordMethod;

    private final AtomicLong _unacknowledgedCount = new AtomicLong(0);
    private final AtomicLong _unacknowledgedBytes = new AtomicLong(0);
    private Subscription _subscription;


    static final class BrowserSubscription extends SubscriptionTarget_0_8
    {
        public BrowserSubscription(AMQChannel channel,
                                   AMQShortString consumerTag, FieldTable filters,
                                   FlowCreditManager creditManager,
                                   ClientDeliveryMethod deliveryMethod,
                                   RecordDeliveryMethod recordMethod)
            throws AMQException
        {
            super(channel, consumerTag,
                  filters, creditManager, deliveryMethod, recordMethod);
        }

        /**
         * This method can be called by each of the publisher threads. As a result all changes to the channel object must be
         * thread safe.
         *
         *
         * @param entry
         * @param batch
         * @throws org.apache.qpid.AMQException
         */
        @Override
        public void send(QueueEntry entry, boolean batch) throws AMQException
        {
            // We don't decrement the reference here as we don't want to consume the message
            // but we do want to send it to the client.

            synchronized (getChannel())
            {
                long deliveryTag = getChannel().getNextDeliveryTag();
                sendToClient(entry.getMessage(), entry.getInstanceProperties(), deliveryTag);
            }

        }

        @Override
        public boolean allocateCredit(QueueEntry msg)
        {
            return true;
        }

    }

    public static class NoAckSubscription extends SubscriptionTarget_0_8
    {
        private final AutoCommitTransaction _txn;

        public NoAckSubscription(AMQChannel channel,
                                 AMQShortString consumerTag, FieldTable filters,
                                 FlowCreditManager creditManager,
                                 ClientDeliveryMethod deliveryMethod,
                                 RecordDeliveryMethod recordMethod)
            throws AMQException
        {
            super(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod);

            _txn = new AutoCommitTransaction(channel.getVirtualHost().getMessageStore());
        }

        /**
         * This method can be called by each of the publisher threads. As a result all changes to the channel object must be
         * thread safe.
         *
         *
         * @param entry   The message to send
         * @param batch
         * @throws org.apache.qpid.AMQException
         */
        @Override
        public void send(QueueEntry entry, boolean batch) throws AMQException
        {
            // if we do not need to wait for client acknowledgements
            // we can decrement the reference count immediately.

            // By doing this _before_ the send we ensure that it
            // doesn't get sent if it can't be dequeued, preventing
            // duplicate delivery on recovery.

            // The send may of course still fail, in which case, as
            // the message is unacked, it will be lost.
            _txn.dequeue(getSubscription().getQueue(), entry.getMessage(), NOOP);

            ServerMessage message = entry.getMessage();
            MessageReference ref = message.newReference();
            InstanceProperties props = entry.getInstanceProperties();
            entry.delete();

            synchronized (getChannel())
            {
                getChannel().getProtocolSession().setDeferFlush(batch);
                long deliveryTag = getChannel().getNextDeliveryTag();

                sendToClient(message, props, deliveryTag);

            }
            ref.release();


        }

        @Override
        public boolean allocateCredit(QueueEntry msg)
        {
            return true;
        }

        private static final ServerTransaction.Action NOOP =
                new ServerTransaction.Action()
                {
                    @Override
                    public void postCommit()
                    {
                    }

                    @Override
                    public void onRollback()
                    {
                    }
                };
    }

    /**
     * NoAck Subscription for use with BasicGet method.
     */
    public static final class GetNoAckSubscription extends SubscriptionTarget_0_8.NoAckSubscription
    {
        public GetNoAckSubscription(AMQChannel channel, AMQProtocolSession protocolSession,
                                    AMQShortString consumerTag, FieldTable filters,
                                    boolean noLocal, FlowCreditManager creditManager,
                                    ClientDeliveryMethod deliveryMethod,
                                    RecordDeliveryMethod recordMethod)
            throws AMQException
        {
            super(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod);
        }

        public boolean allocateCredit(QueueEntry msg)
        {
            return getCreditManager().useCreditForMessage(msg.getMessage().getSize());
        }

    }

    static final class AckSubscription extends SubscriptionTarget_0_8
    {
        public AckSubscription(AMQChannel channel,
                               AMQShortString consumerTag, FieldTable filters,
                               FlowCreditManager creditManager,
                               ClientDeliveryMethod deliveryMethod,
                               RecordDeliveryMethod recordMethod)
            throws AMQException
        {
            super(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod);
        }

        /**
         * This method can be called by each of the publisher threads. As a result all changes to the channel object must be
         * thread safe.
         *
         *
         * @param entry   The message to send
         * @param batch
         * @throws org.apache.qpid.AMQException
         */
        @Override
        public void send(QueueEntry entry, boolean batch) throws AMQException
        {


            synchronized (getChannel())
            {
                getChannel().getProtocolSession().setDeferFlush(batch);
                long deliveryTag = getChannel().getNextDeliveryTag();

                addUnacknowledgedMessage(entry);
                recordMessageDelivery(entry, deliveryTag);
                entry.addStateChangeListener(getReleasedStateChangeListener());
                sendToClient(entry.getMessage(), entry.getInstanceProperties(), deliveryTag);
                entry.incrementDeliveryCount();

            }
        }



    }


    private static final Logger _logger = Logger.getLogger(SubscriptionTarget_0_8.class);

    private final AMQChannel _channel;

    private final AMQShortString _consumerTag;

    private final FlowCreditManager _creditManager;

    private final Boolean _autoClose;

    private final AtomicBoolean _deleted = new AtomicBoolean(false);




    public SubscriptionTarget_0_8(AMQChannel channel,
                                  AMQShortString consumerTag,
                                  FieldTable arguments,
                                  FlowCreditManager creditManager,
                                  ClientDeliveryMethod deliveryMethod,
                                  RecordDeliveryMethod recordMethod)
            throws AMQException
    {
        super(State.ACTIVE);

        _channel = channel;
        _consumerTag = consumerTag;

        _creditManager = creditManager;
        creditManager.addStateListener(this);

        _deliveryMethod = deliveryMethod;
        _recordMethod = recordMethod;

        if (arguments != null)
        {
            Object autoClose = arguments.get(AMQPFilterTypes.AUTO_CLOSE.getValue());
            if (autoClose != null)
            {
                _autoClose = (Boolean) autoClose;
            }
            else
            {
                _autoClose = false;
            }
        }
        else
        {
            _autoClose = false;
        }
    }

    public void setSubscription(Subscription subscription)
    {
        _subscription = subscription;
    }

    public Subscription getSubscription()
    {
        return _subscription;
    }

    public AMQSessionModel getSessionModel()
    {
        return _channel;
    }

    public String toString()
    {
        String subscriber = "[channel=" + _channel +
                            ", consumerTag=" + _consumerTag +
                            ", session=" + getProtocolSession().getKey()  ;

        return subscriber + "]";
    }

    public boolean isSuspended()
    {
        return getState()!=State.ACTIVE || _channel.isSuspended() || _deleted.get() || _channel.getConnectionModel().isStopped();
    }

    /**
     * Callback indicating that a queue has been deleted.
     *
     */
    public void queueDeleted()
    {
        _deleted.set(true);
    }

    public boolean isAutoClose()
    {
        return _autoClose;
    }

    public FlowCreditManager getCreditManager()
    {
        return _creditManager;
    }


    public boolean close()
    {
        boolean closed = false;
        State state = getState();

        getSubscription().getSendLock();
        try
        {
            while(!closed && state != State.CLOSED)
            {
                closed = updateState(state, State.CLOSED);
                if(!closed)
                {
                    state = getState();
                }
            }
            _creditManager.removeListener(this);
            return closed;
        }
        finally
        {
            getSubscription().releaseSendLock();
        }
    }


    public boolean allocateCredit(QueueEntry msg)
    {
        return _creditManager.useCreditForMessage(msg.getMessage().getSize());
    }

    public AMQChannel getChannel()
    {
        return _channel;
    }

    public AMQShortString getConsumerTag()
    {
        return _consumerTag;
    }

    public AMQProtocolSession getProtocolSession()
    {
        return _channel.getProtocolSession();
    }

    public void restoreCredit(final QueueEntry queueEntry)
    {
        _creditManager.restoreCredit(1, queueEntry.getSize());
    }

    protected final StateChangeListener<QueueEntry, QueueEntry.State> getReleasedStateChangeListener()
    {
        return _entryReleaseListener;
    }

    public void creditStateChanged(boolean hasCredit)
    {

        if(hasCredit)
        {
            if(!updateState(State.SUSPENDED, State.ACTIVE))
            {
                // this is a hack to get round the issue of increasing bytes credit
                getStateListener().stateChanged(this, State.ACTIVE, State.ACTIVE);
            }
        }
        else
        {
            updateState(State.ACTIVE, State.SUSPENDED);
        }
    }

    protected void sendToClient(final ServerMessage message, final InstanceProperties props, final long deliveryTag)
            throws AMQException
    {
        _deliveryMethod.deliverToClient(getSubscription(), message, props, deliveryTag);

    }


    protected void recordMessageDelivery(final QueueEntry entry, final long deliveryTag)
    {
        _recordMethod.recordMessageDelivery(getSubscription(),entry,deliveryTag);
    }


    public void confirmAutoClose()
    {
        ProtocolOutputConverter converter = getChannel().getProtocolSession().getProtocolOutputConverter();
        converter.confirmConsumerAutoClose(getChannel().getChannelId(), getConsumerTag());
    }

    public void queueEmpty() throws AMQException
    {
        if (isAutoClose())
        {
            close();
            confirmAutoClose();
        }
    }

    public void flushBatched()
    {
        _channel.getProtocolSession().setDeferFlush(false);

        _channel.getProtocolSession().flushBatched();
    }

    protected void addUnacknowledgedMessage(QueueEntry entry)
    {
        final long size = entry.getSize();
        _unacknowledgedBytes.addAndGet(size);
        _unacknowledgedCount.incrementAndGet();
        entry.addStateChangeListener(new StateChangeListener<QueueEntry, QueueEntry.State>()
        {
            public void stateChanged(QueueEntry entry, QueueEntry.State oldState, QueueEntry.State newState)
            {
                if(oldState.equals(QueueEntry.State.ACQUIRED) && !newState.equals(QueueEntry.State.ACQUIRED))
                {
                    _unacknowledgedBytes.addAndGet(-size);
                    _unacknowledgedCount.decrementAndGet();
                    entry.removeStateChangeListener(this);
                }
            }
        });
    }

    public long getUnacknowledgedBytes()
    {
        return _unacknowledgedBytes.longValue();
    }

    public long getUnacknowledgedMessages()
    {
        return _unacknowledgedCount.longValue();
    }
}