content: /*
   * 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.activemq.transport.amqp.client;
  
  import java.io.IOException;
  import java.util.Map;
  import java.util.concurrent.ScheduledExecutorService;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicLong;
  
  import org.apache.activemq.transport.amqp.client.util.AsyncResult;
  import org.apache.activemq.transport.amqp.client.util.ClientFuture;
  import org.apache.activemq.transport.amqp.client.util.UnmodifiableProxy;
  import org.apache.qpid.proton.amqp.Symbol;
  import org.apache.qpid.proton.amqp.messaging.Source;
  import org.apache.qpid.proton.amqp.messaging.Target;
  import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode;
  import org.apache.qpid.proton.amqp.transport.SenderSettleMode;
  import org.apache.qpid.proton.engine.Connection;
  import org.apache.qpid.proton.engine.Session;
  
  /**
   * Session class that manages a Proton session endpoint.
   */
  public class AmqpSession extends AmqpAbstractResource<Session> {
  
     private final AtomicLong receiverIdGenerator = new AtomicLong();
     private final AtomicLong senderIdGenerator = new AtomicLong();
  
     private final AmqpConnection connection;
     private final String sessionId;
     private final AmqpTransactionContext txContext;
     private final AtomicBoolean closed = new AtomicBoolean();
  
     /**
      * Create a new session instance.
      *
      * @param connection The parent connection that created the session.
      * @param sessionId  The unique ID value assigned to this session.
      */
     public AmqpSession(AmqpConnection connection, String sessionId) {
        this.connection = connection;
        this.sessionId = sessionId;
        this.txContext = new AmqpTransactionContext(this);
     }
  
     /**
      * Close the receiver, a closed receiver will throw exceptions if any further send calls are made.
      *
      * @throws IOException if an error occurs while closing the receiver.
      */
     public void close() throws IOException {
        if (closed.compareAndSet(false, true)) {
           final ClientFuture request = new ClientFuture();
           getScheduler().execute(() -> {
              checkClosed();
              close(request);
              pumpToProtonTransport(request);
           });
  
           request.sync();
        }
     }
  
     /**
      * Create an anonymous sender.
      *
      * @return a newly created sender that is ready for use
      * @throws Exception if an error occurs while creating the sender.
      */
     public AmqpSender createSender() throws Exception {
        return createSender(null, false);
     }
  
     /**
      * Create an anonymous sender instance using the anonymous relay support of the broker.
      *
      * @return a newly created sender that is ready for use
      * @throws Exception if an error occurs while creating the sender.
      */
     public AmqpSender createAnonymousSender() throws Exception {
        return createSender(null, false);
     }
  
     /**
      * Create a sender instance using the given address
      *
      * @param address the address to which the sender will produce its messages.
      * @return a newly created sender that is ready for use
      * @throws Exception if an error occurs while creating the sender.
      */
     public AmqpSender createSender(final String address) throws Exception {
        return createSender(address, false);
     }
  
     /**
      * Create a sender instance using the given address
      *
      * @param address             the address to which the sender will produce its messages.
      * @param desiredCapabilities the capabilities that the caller wants the remote to support.
      * @return a newly created sender that is ready for use
      * @throws Exception if an error occurs while creating the sender.
      */
     public AmqpSender createSender(final String address, Symbol[] desiredCapabilities) throws Exception {
        return createSender(address, false, desiredCapabilities, null, null);
     }
  
     /**
      * Create a sender instance using the given address
      *
      * @param address   the address to which the sender will produce its messages.
      * @param presettle controls if the created sender produces message that have already been marked settled.
      * @return a newly created sender that is ready for use
      * @throws Exception if an error occurs while creating the sender.
      */
     public AmqpSender createSender(final String address, boolean presettle) throws Exception {
        return createSender(address, presettle, null, null, null);
     }
  
     /**
      * Create a sender instance using the given address
      *
      * @param address             the address to which the sender will produce its messages.
      * @param presettle           controls if the created sender produces message that have already been marked settled.
      * @param desiredCapabilities the capabilities that the caller wants the remote to support.
      * @param offeredCapabilities the capabilities that the caller wants the advertise support for.
      * @param properties          the properties to send as part of the sender open.
      * @return a newly created sender that is ready for use
      * @throws Exception if an error occurs while creating the sender.
      */
     public AmqpSender createSender(final String address, boolean presettle, Symbol[] desiredCapabilities, Symbol[] offeredCapabilities, Map<Symbol, Object> properties) throws Exception {
        checkClosed();
  
        final AmqpSender sender = new AmqpSender(AmqpSession.this, address, getNextSenderId());
        sender.setPresettle(presettle);
        sender.setDesiredCapabilities(desiredCapabilities);
        sender.setOfferedCapabilities(offeredCapabilities);
        sender.setProperties(properties);
        final ClientFuture request = new ClientFuture();
  
        connection.getScheduler().execute(() -> {
           checkClosed();
           sender.setStateInspector(getStateInspector());
           sender.open(request);
           pumpToProtonTransport(request);
        });
  
        request.sync();
  
        return sender;
     }
  
     /**
      * Create a sender instance using the given address
      *
      * @param address      the address to which the sender will produce its messages.
      * @param senderMode   controls the settlement mode used by the created Sender
      * @param receiverMode controls the desired settlement mode used by the remote Receiver
      * @return a newly created sender that is ready for use
      * @throws Exception if an error occurs while creating the sender.
      */
     public AmqpSender createSender(final String address,
                                    final SenderSettleMode senderMode,
                                    ReceiverSettleMode receiverMode) throws Exception {
        return createSender(address, senderMode, receiverMode, AmqpSender.DEFAULT_OUTCOMES);
     }
  
     /**
      * Create a sender instance using the given address
      *
      * @param address      the address to which the sender will produce its messages.
      * @param senderMode   controls the settlement mode used by the created Sender
      * @param receiverMode controls the desired settlement mode used by the remote Receiver
      * @param outcomes     specifies the outcomes supported by the sender
      * @return a newly created sender that is ready for use
      * @throws Exception if an error occurs while creating the sender.
      */
     public AmqpSender createSender(final String address,
                                    final SenderSettleMode senderMode,
                                    ReceiverSettleMode receiverMode, final Symbol[] outcomes) throws Exception {
        checkClosed();
  
        final AmqpSender sender = new AmqpSender(AmqpSession.this, address, getNextSenderId(), senderMode, receiverMode, outcomes);
        final ClientFuture request = new ClientFuture();
  
        connection.getScheduler().execute(() -> {
           checkClosed();
           sender.setStateInspector(getStateInspector());
           sender.open(request);
           pumpToProtonTransport(request);
        });
  
        request.sync();
  
        return sender;
     }
  
     /**
      * Create a sender instance using the given Target
      *
      * @param target the caller created and configured Target used to create the sender link.
      * @return a newly created sender that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpSender createSender(Target target) throws Exception {
        return createSender(target, getNextSenderId());
     }
  
     /**
      * Create a sender instance using the given Target
      *
      * @param target   the caller created and configured Target used to create the sender link.
      * @param senderId the sender ID to assign to the newly created Sender.
      * @return a newly created sender that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpSender createSender(Target target, String senderId) throws Exception {
        return createSender(target, senderId, null, null, null);
     }
  
     /**
      * Create a sender instance using the given Target
      *
      * @param target              the caller created and configured Target used to create the sender link.
      * @param senderId            the sender ID to assign to the newly created Sender.
      * @param desiredCapabilities the capabilities that the caller wants the remote to support.
      * @param offeredCapabilities the capabilities that the caller wants the advertise support for.
      * @param properties          the properties to send as part of the sender open.
      * @return a newly created sender that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpSender createSender(Target target, String senderId, Symbol[] desiredCapabilities, Symbol[] offeredCapabilities, Map<Symbol, Object> properties) throws Exception {
        checkClosed();
  
        final AmqpSender sender = new AmqpSender(AmqpSession.this, target, senderId);
        sender.setDesiredCapabilities(desiredCapabilities);
        sender.setOfferedCapabilities(offeredCapabilities);
        sender.setProperties(properties);
        final ClientFuture request = new ClientFuture();
  
        connection.getScheduler().execute(() -> {
           checkClosed();
           sender.setStateInspector(getStateInspector());
           sender.open(request);
           pumpToProtonTransport(request);
        });
  
        request.sync();
  
        return sender;
     }
  
     /**
      * Create a receiver instance using the given address
      *
      * @param address the address to which the receiver will subscribe for its messages.
      * @return a newly created receiver that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createReceiver(String address) throws Exception {
        return createReceiver(address, null, false);
     }
  
     /**
      * Create a receiver instance using the given address
      *
      * @param address  the address to which the receiver will subscribe for its messages.
      * @param selector the JMS selector to use for the subscription
      * @return a newly created receiver that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createReceiver(String address, String selector) throws Exception {
        return createReceiver(address, selector, false);
     }
  
     /**
      * Create a receiver instance using the given address
      *
      * @param address  the address to which the receiver will subscribe for its messages.
      * @param selector the JMS selector to use for the subscription
      * @param noLocal  should the subscription have messages from its connection filtered.
      * @return a newly created receiver that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createReceiver(String address, String selector, boolean noLocal) throws Exception {
        return createReceiver(address, selector, noLocal, false);
     }
  
     public AmqpReceiver createReceiver(String address,
                                        String selector,
                                        boolean noLocal,
                                        boolean presettle) throws Exception {
        return createReceiver(address, selector, noLocal, presettle, null);
     }
  
     /**
      * Create a receiver instance using the given address
      *
      * @param address    the address to which the receiver will subscribe for its messages.
      * @param selector   the JMS selector to use for the subscription
      * @param noLocal    should the subscription have messages from its connection filtered.
      * @param presettle  should the receiver be created with a settled sender mode.
      * @param properties to set on the receiver
      * @return a newly created receiver that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createReceiver(String address,
                                        String selector,
                                        boolean noLocal,
                                        boolean presettle,
                                        Map<Symbol, Object> properties) throws Exception {
        checkClosed();
  
        final ClientFuture request = new ClientFuture();
        final AmqpReceiver receiver = new AmqpReceiver(AmqpSession.this, address, getNextReceiverId());
  
        receiver.setNoLocal(noLocal);
        receiver.setPresettle(presettle);
        if (selector != null && !selector.isEmpty()) {
           receiver.setSelector(selector);
        }
        if (properties != null && !properties.isEmpty()) {
           receiver.setProperties(properties);
        }
  
        connection.getScheduler().execute(() -> {
           checkClosed();
           receiver.setStateInspector(getStateInspector());
           receiver.open(request);
           pumpToProtonTransport(request);
        });
  
        request.sync();
  
        return receiver;
     }
  
     /**
      * Create a receiver instance using the given address
      *
      * @param address      the address to which the receiver will subscribe for its messages.
      * @param senderMode   controls the desired settlement mode used by the remote Sender
      * @param receiverMode controls the settlement mode used by the created Receiver
      * @return a newly created receiver that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createReceiver(String address, SenderSettleMode senderMode, ReceiverSettleMode receiverMode) throws Exception {
        checkClosed();
  
        final ClientFuture request = new ClientFuture();
        final AmqpReceiver receiver = new AmqpReceiver(AmqpSession.this, address, getNextReceiverId(), senderMode, receiverMode);
  
        connection.getScheduler().execute(() -> {
           checkClosed();
           receiver.setStateInspector(getStateInspector());
           receiver.open(request);
           pumpToProtonTransport(request);
        });
  
        request.sync();
  
        return receiver;
     }
  
     /**
      * Create a receiver instance using the given Source
      *
      * @param source the caller created and configured Source used to create the receiver link.
      * @return a newly created receiver that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createReceiver(Source source) throws Exception {
        return createReceiver(source, getNextReceiverId());
     }
  
     /**
      * Create a receiver instance using the given Source
      *
      * @param source     the caller created and configured Source used to create the receiver link.
      * @param receiverId the receiver id to use.
      * @return a newly created receiver that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createReceiver(Source source, String receiverId) throws Exception {
        checkClosed();
  
        final ClientFuture request = new ClientFuture();
        final AmqpReceiver receiver = new AmqpReceiver(AmqpSession.this, source, receiverId);
  
        connection.getScheduler().execute(() -> {
           checkClosed();
           receiver.setStateInspector(getStateInspector());
           receiver.open(request);
           pumpToProtonTransport(request);
        });
  
        request.sync();
  
        return receiver;
     }
  
     /**
      * Create a receiver instance using the given Source
      *
      * @param source the caller created and configured Source used to create the receiver link.
      * @return a newly created receiver that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createMulticastReceiver(Source source, String receiverId, String receiveName) throws Exception {
        checkClosed();
  
        final ClientFuture request = new ClientFuture();
        final AmqpReceiver receiver = new AmqpReceiver(AmqpSession.this, source, receiverId);
        receiver.setSubscriptionName(receiveName);
  
        connection.getScheduler().execute(() -> {
           checkClosed();
           receiver.setStateInspector(getStateInspector());
           receiver.open(request);
           pumpToProtonTransport(request);
        });
  
        request.sync();
  
        return receiver;
     }
  
     /**
      * Create a receiver instance using the given receiverId
      *
      * @return a newly created receiver that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createMulticastReceiver(String receiverId, String address, String receiveName) throws Exception {
        checkClosed();
  
        final ClientFuture request = new ClientFuture();
        final AmqpReceiver receiver = new AmqpReceiver(AmqpSession.this, address, receiverId);
        receiver.setSubscriptionName(receiveName);
  
        connection.getScheduler().execute(() -> {
           checkClosed();
           receiver.setStateInspector(getStateInspector());
           receiver.open(request);
           pumpToProtonTransport(request);
        });
  
        request.sync();
  
        return receiver;
     }
  
     /**
      * Create a receiver instance using the given address that creates a durable subscription.
      *
      * @param address          the address to which the receiver will subscribe for its messages.
      * @param subscriptionName the name of the subscription that is being created.
      * @return a newly created receiver that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createDurableReceiver(String address, String subscriptionName) throws Exception {
        return createDurableReceiver(address, subscriptionName, null, false);
     }
  
     /**
      * Create a receiver instance using the given address that creates a durable subscription.
      *
      * @param address          the address to which the receiver will subscribe for its messages.
      * @param subscriptionName the name of the subscription that is being created.
      * @param selector         the JMS selector to use for the subscription
      * @return a newly created receiver that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createDurableReceiver(String address,
                                               String subscriptionName,
                                               String selector) throws Exception {
        return createDurableReceiver(address, subscriptionName, selector, false);
     }
  
     /**
      * Create a receiver instance using the given address that creates a durable subscription.
      *
      * @param address          the address to which the receiver will subscribe for its messages.
      * @param subscriptionName the name of the subscription that is being created.
      * @param selector         the JMS selector to use for the subscription
      * @param noLocal          should the subscription have messages from its connection filtered.
      * @return a newly created receiver that is ready for use
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createDurableReceiver(String address,
                                               String subscriptionName,
                                               String selector,
                                               boolean noLocal) throws Exception {
        checkClosed();
  
        if (subscriptionName == null || subscriptionName.isEmpty()) {
           throw new IllegalArgumentException("subscription name must not be null or empty.");
        }
  
        final ClientFuture request = new ClientFuture();
        final AmqpReceiver receiver = new AmqpReceiver(AmqpSession.this, address, getNextReceiverId());
        receiver.setSubscriptionName(subscriptionName);
        receiver.setNoLocal(noLocal);
        if (selector != null && !selector.isEmpty()) {
           receiver.setSelector(selector);
        }
  
        connection.getScheduler().execute(() -> {
           checkClosed();
           receiver.setStateInspector(getStateInspector());
           receiver.open(request);
           pumpToProtonTransport(request);
        });
  
        request.sync();
  
        return receiver;
     }
  
     /**
      * Create a receiver instance using the given address that creates a durable subscription.
      *
      * @param subscriptionName the name of the subscription that should be queried for on the remote..
      * @return a newly created receiver that is ready for use if the subscription exists
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver lookupSubscription(String subscriptionName) throws Exception {
        checkClosed();
  
        if (subscriptionName == null || subscriptionName.isEmpty()) {
           throw new IllegalArgumentException("subscription name must not be null or empty.");
        }
  
        final ClientFuture request = new ClientFuture();
        final AmqpReceiver receiver = new AmqpReceiver(AmqpSession.this, (String) null, getNextReceiverId());
        receiver.setSubscriptionName(subscriptionName);
  
        connection.getScheduler().execute(() -> {
           checkClosed();
           receiver.setStateInspector(getStateInspector());
           receiver.open(request);
           pumpToProtonTransport(request);
        });
  
        request.sync();
  
        return receiver;
     }
  
     /**
      * {@return this session's parent AmqpConnection}
      */
     public AmqpConnection getConnection() {
        return connection;
     }
  
     public Session getSession() {
        return UnmodifiableProxy.sessionProxy(getEndpoint());
     }
  
     public boolean isInTransaction() {
        return txContext.isInTransaction();
     }
  
     @Override
     public String toString() {
        return "AmqpSession { " + sessionId + " }";
     }
  
     //----- Session Transaction Methods --------------------------------------//
  
     /**
      * Starts a new transaction associated with this session.
      *
      * @throws Exception if an error occurs starting a new Transaction.
      */
     public void begin() throws Exception {
        if (txContext.isInTransaction()) {
           throw new javax.jms.IllegalStateException("Session already has an active transaction");
        }
  
        txContext.begin();
     }
  
     /**
      * Commit the current transaction associated with this session.
      *
      * @throws Exception if an error occurs committing the Transaction.
      */
     public void commit() throws Exception {
        if (!txContext.isInTransaction()) {
           throw new javax.jms.IllegalStateException("Commit called on Session that does not have an active transaction");
        }
  
        txContext.commit();
     }
  
     /**
      * Roll back the current transaction associated with this session.
      *
      * @throws Exception if an error occurs rolling back the Transaction.
      */
     public void rollback() throws Exception {
        if (!txContext.isInTransaction()) {
           throw new javax.jms.IllegalStateException("Rollback called on Session that does not have an active transaction");
        }
  
        txContext.rollback();
     }
  
     //----- Internal access used to manage resources -------------------------//
  
     ScheduledExecutorService getScheduler() {
        return connection.getScheduler();
     }
  
     Connection getProtonConnection() {
        return connection.getProtonConnection();
     }
  
     void pumpToProtonTransport(AsyncResult request) {
        connection.pumpToProtonTransport(request);
     }
  
     public AmqpTransactionId getTransactionId() {
        if (txContext != null && txContext.isInTransaction()) {
           return txContext.getTransactionId();
        }
  
        return null;
     }
  
     AmqpTransactionContext getTransactionContext() {
        return txContext;
     }
  
     //----- Private implementation details -----------------------------------//
  
     @Override
     protected void doOpenInspection() {
        try {
           getStateInspector().inspectOpenedResource(getSession());
        } catch (Throwable error) {
           getStateInspector().markAsInvalid(error.getMessage());
        }
     }
  
     @Override
     protected void doClosedInspection() {
        try {
           getStateInspector().inspectClosedResource(getSession());
        } catch (Throwable error) {
           getStateInspector().markAsInvalid(error.getMessage());
        }
     }
  
     private String getNextSenderId() {
        return sessionId + ":" + senderIdGenerator.incrementAndGet();
     }
  
     private String getNextReceiverId() {
        return sessionId + ":" + receiverIdGenerator.incrementAndGet();
     }
  
     private void checkClosed() {
        if (isClosed() || connection.isClosed()) {
           throw new IllegalStateException("Session is already closed");
        }
     }
  }