broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Session_0_10.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_10;

import java.security.AccessControlContext;
import java.util.Collection;
import java.util.List;

import javax.security.auth.Subject;

import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.model.Connection;
import org.apache.qpid.server.model.Consumer;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.Session;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.ConsumerListener;
import org.apache.qpid.server.protocol.PublishAuthorisationCache;
import org.apache.qpid.server.session.AbstractAMQPSession;
import org.apache.qpid.server.util.Action;

public class Session_0_10 extends AbstractAMQPSession<Session_0_10, ConsumerTarget_0_10>
        implements AMQSessionModel<Session_0_10, ConsumerTarget_0_10>, LogSubject
{
    private final AMQPConnection_0_10 _connection;
    private final ServerSession _serverSession;

    protected Session_0_10(final Connection<?> parent, final int sessionId, final ServerSession serverSession)
    {
        super(parent, sessionId);
        _connection = (AMQPConnection_0_10) parent;
        _serverSession = serverSession;
    }

    @Override
    public String toLogString()
    {
        return _serverSession.toLogString();
    }

    @Override
    public void block(final Queue<?> queue)
    {
        _serverSession.block(queue);
    }

    @Override
    public void unblock(final Queue<?> queue)
    {
        _serverSession.unblock(queue);
    }

    @Override
    public void block()
    {
        _serverSession.block();
    }

    @Override
    public void unblock()
    {
        _serverSession.unblock();
    }

    @Override
    public Object getConnectionReference()
    {
        return _serverSession.getConnectionReference();
    }

    @Override
    public void addConsumerListener(final ConsumerListener listener)
    {
        _serverSession.addConsumerListener(listener);
    }

    @Override
    public void removeConsumerListener(final ConsumerListener listener)
    {
        _serverSession.removeConsumerListener(listener);
    }

    @Override
    public void setModelObject(final Session<?> session)
    {
        _serverSession.setModelObject(this);
    }

    @Override
    public Session<?> getModelObject()
    {
        return this;
    }

    @Override
    public void transportStateChanged()
    {
        for(ConsumerTarget_0_10 consumerTarget : _serverSession.getSubscriptions())
        {
            consumerTarget.transportStateChanged();
        }
        if (!_consumersWithPendingWork.isEmpty() && !getAMQPConnection().isTransportBlockedForWriting())
        {
            getAMQPConnection().notifyWork(this);
        }
    }

    @Override
    protected void updateBlockedStateIfNecessary()
    {
        _serverSession.updateBlockedStateIfNecesssary();
    }

    @Override
    public int compareTo(final AMQSessionModel o)
    {
        return getId().compareTo(o.getId());
    }

    @Override
    public boolean getBlocking()
    {
        return _serverSession.getBlocking();
    }

    @Override
    public Collection<Consumer<?, ConsumerTarget_0_10>> getConsumers()
    {
        return _serverSession.getConsumers();
    }

    @Override
    public long getConsumerCount()
    {
        return _serverSession.getConsumerCount();
    }

    @Override
    public long getTxnRejects()
    {
        return _serverSession.getTxnRejects();
    }

    @Override
    public long getTxnCommits()
    {
        return _serverSession.getTxnCommits();
    }

    @Override
    public long getTxnStart()
    {
        return _serverSession.getTxnStart();
    }

    @Override
    public int getUnacknowledgedMessageCount()
    {
        return _serverSession.getUnacknowledgedMessageCount();
    }

    @Override
    public void doTimeoutAction(final String idleTransactionTimeoutError)
    {
        _serverSession.doTimeoutAction(idleTransactionTimeoutError);
    }

    @Override
    public long getTransactionUpdateTimeLong()
    {
        return _serverSession.getTransactionUpdateTimeLong();
    }

    @Override
    public long getTransactionStartTimeLong()
    {
        return _serverSession.getTransactionStartTimeLong();
    }

    public AMQPConnection_0_10 getConnection()
    {
        return _connection;
    }

    public Subject getSubject()
    {
        return _subject;
    }

    public AccessControlContext getAccessControllerContext()
    {
        return _accessControllerContext;
    }

    public PublishAuthorisationCache getPublishAuthCache()
    {
        return _publishAuthCache;
    }

    public List<Action<? super Session_0_10>> getTaskList()
    {
        return _taskList;
    }

    public boolean isClosing()
    {
        return _serverSession.isClosing();
    }

    public ServerSession getServerSession()
    {
        return _serverSession;
    }

    public long getMaxUncommittedInMemorySize()
    {
        return _maxUncommittedInMemorySize;
    }
}