/**
 * 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(new Runnable() {

      @Override
      public void run() {
         checkClosed();
         sender.setStateInspector(getStateInspector());
         sender.open(request);
         pumpToProtonTransport(request);
      }
   });

   request.sync();

   return sender;
}
 

@Test
public void testGetSendBufferAddsDeliveryCountOnlyToSendMessageAndTrimsDeliveryAnnotations() {
   MessageImpl protonMessage = createProtonMessage();
   DeliveryAnnotations deliveryAnnotations = new DeliveryAnnotations(new HashMap<>());
   deliveryAnnotations.getValue().put(Symbol.valueOf("testGetSendBufferRemoveDeliveryAnnotations"), "X");
   protonMessage.setDeliveryAnnotations(deliveryAnnotations);
   AMQPStandardMessage message = new AMQPStandardMessage(0, encodeMessage(protonMessage), null, null);

   ReadableBuffer buffer = message.getSendBuffer(7);
   assertNotNull(buffer);
   message.reencode(); // Ensures Header is current if accidentally updated

   AMQPStandardMessage copy = new AMQPStandardMessage(0, buffer, null, null);

   MessageImpl originalsProtonMessage = message.getProtonMessage();
   MessageImpl copyProtonMessage = copy.getProtonMessage();
   assertProtonMessageNotEquals(originalsProtonMessage, copyProtonMessage);

   assertNull(originalsProtonMessage.getHeader().getDeliveryCount());
   assertEquals(6, copyProtonMessage.getHeader().getDeliveryCount().intValue());
   assertNull(copyProtonMessage.getDeliveryAnnotations());
}
 

private void doTestEncodeSymbolTypeReservation(int size) throws IOException {
    Random random = new Random(System.currentTimeMillis());
    StringBuilder builder = new StringBuilder(size);
    for (int i = 0; i < size; ++i) {
        builder.append((byte) random.nextInt(127));
    }

    Symbol symbol = Symbol.valueOf(builder.toString());

    WritableBuffer writable = new WritableBuffer.ByteBufferWrapper(ByteBuffer.allocate(2048));
    WritableBuffer spy = Mockito.spy(writable);

    encoder.setByteBuffer(spy);
    encoder.writeSymbol(symbol);

    // Check that the SymbolType tries to reserve space, actual encoding size not computed here.
    Mockito.verify(spy).ensureRemaining(Mockito.anyInt());
}
 

@Test
public void testConditionAndDescriptionEquality()
{
    String symbolValue = "symbol";
    String descriptionValue = "description";

    ErrorCondition same1 = new ErrorCondition();
    same1.setCondition(Symbol.getSymbol(new String(symbolValue)));
    same1.setDescription(new String(descriptionValue));

    ErrorCondition same2 = new ErrorCondition();
    same2.setCondition(Symbol.getSymbol(new String(symbolValue)));
    same2.setDescription(new String(descriptionValue));

    assertErrorConditionsEqual(same1, same2);

    ErrorCondition different = new ErrorCondition();
    different.setCondition(Symbol.getSymbol(symbolValue));
    different.setDescription("other");

    assertErrorConditionsNotEqual(same1, different);
}
 

@Test
public void testSetMessageAnnotationsOnNewMessage() {
    Symbol symbolKeyName = Symbol.valueOf("myTestSymbolName");
    Symbol symbolKeyName2 = Symbol.valueOf("myTestSymbolName2");
    String value = "myTestValue";

    AmqpJmsMessageFacade amqpMessageFacade = createNewMessageFacade();

    // check setting first annotation
    amqpMessageFacade.setMessageAnnotation(symbolKeyName, value);

    MessageAnnotations underlyingAnnotations = amqpMessageFacade.getMessageAnnotations();
    assertNotNull(underlyingAnnotations);

    assertTrue(underlyingAnnotations.getValue().containsKey(symbolKeyName));
    assertEquals(value, underlyingAnnotations.getValue().get(symbolKeyName));

    // set another
    amqpMessageFacade.setMessageAnnotation(symbolKeyName2, value);

    assertTrue(underlyingAnnotations.getValue().containsKey(symbolKeyName));
    assertTrue(underlyingAnnotations.getValue().containsKey(symbolKeyName2));
}
 

@Test
public void testEncodeDecodeSymbolArrayUsingPutObject()
{
    Symbol symbol1 = Symbol.valueOf("testRoundtripSymbolArray1");
    Symbol symbol2 = Symbol.valueOf("testRoundtripSymbolArray2");
    Symbol[] input = new Symbol[]{symbol1, symbol2};

    Data data1 = new DataImpl();
    data1.putObject(input);

    Binary encoded = data1.encode();
    encoded.asByteBuffer();

    Data data2 = new DataImpl();
    data2.decode(encoded.asByteBuffer());

    assertEquals("unexpected array length", 2, data2.getArray());
    assertEquals("unexpected array length", Data.DataType.SYMBOL, data2.getArrayType());

    Object[] array = data2.getJavaArray();
    assertArrayEquals("Array not as expected", input, array);
}
 

protected FrameWithPayloadMatchingHandler(FrameType frameType,
                                            int channel,
                                            int frameSize,
                                            UnsignedLong numericDescriptor,
                                            Symbol symbolicDescriptor)
{
    super(frameType, channel, frameSize, numericDescriptor, symbolicDescriptor);
}
 

protected AbstractFrameFieldAndPayloadMatchingHandler(FrameType frameType,
                                            int channel,
                                            int frameSize,
                                            UnsignedLong numericDescriptor,
                                            Symbol symbolicDescriptor)
{
    super(numericDescriptor, symbolicDescriptor);
    _frameType = frameType;
    _expectedChannel = channel;
    _expectedFrameSize = frameSize;
}
 

public void expectSaslFailingExchange(Symbol[] serverMechs, Symbol clientSelectedMech, UnsignedByte saslFailureAuthCode)
{
    SaslMechanismsFrame saslMechanismsFrame = new SaslMechanismsFrame().setSaslServerMechanisms(serverMechs);
    addHandler(new HeaderHandlerImpl(AmqpHeader.SASL_HEADER, AmqpHeader.SASL_HEADER,
                                        new FrameSender(
                                                this, FrameType.SASL, 0,
                                                saslMechanismsFrame, null)));

    if(saslFailureAuthCode.compareTo(SASL_FAIL_AUTH) < 0 || saslFailureAuthCode.compareTo(SASL_SYS_TEMP) > 0) {
        throw new IllegalArgumentException("A valid failing SASL code must be supplied");
    }

    SaslInitMatcher saslInitMatcher = new SaslInitMatcher().withMechanism(equalTo(clientSelectedMech));
    saslInitMatcher.onCompletion(new AmqpPeerRunnable()
    {
        @Override
        public void run()
        {
            TestAmqpPeer.this.sendFrame(
                    FrameType.SASL, 0,
                    new SaslOutcomeFrame().setCode(saslFailureAuthCode),
                    null,
                    false, 0);
            _driverRunnable.expectHeader();
        }
    });
    addHandler(saslInitMatcher);
}
 

@Test(timeout = 20000)
public void testRestrictSaslMechanismsWithSingleMech() throws Exception {
    // Check PLAIN gets picked when we don't specify a restriction
    doMechanismSelectionRestrictedTestImpl("username", "password", PLAIN, new Symbol[] { PLAIN, ANONYMOUS}, null);

    // Check ANONYMOUS gets picked when we do specify a restriction
    doMechanismSelectionRestrictedTestImpl("username", "password", ANONYMOUS, new Symbol[] { PLAIN, ANONYMOUS}, "ANONYMOUS");
}
 

public SaslInitMatcher()
{
    super(FrameType.SASL,
          ANY_CHANNEL,
          UnsignedLong.valueOf(0x0000000000000041L),
          Symbol.valueOf("amqp:sasl-init:list"));
}
 

private static byte parseQueueAnnotation(MessageAnnotations annotations, Symbol symbol) {
   Object value = (annotations != null && annotations.getValue() != null ? annotations.getValue().get(symbol) : AMQPMessageSupport.QUEUE_TYPE);

   byte queueType;
   if (value == null || !(value instanceof Number)) {
      queueType = AMQPMessageSupport.QUEUE_TYPE;
   } else {
      queueType = ((Number)value).byteValue();
   }
   return queueType;
}
 

@Override
protected Receiver createEndpoint(JmsSessionInfo resourceInfo) {
    Receiver receiver = getParent().getEndpoint().receiver(linkName);
    receiver.setTarget(new Target());
    receiver.setSenderSettleMode(SenderSettleMode.UNSETTLED);
    receiver.setReceiverSettleMode(ReceiverSettleMode.FIRST);

    if (!hasClientID) {
      // We are trying to unsubscribe a 'global' shared subs using a 'null source lookup', add link
      // desired capabilities as hints to the peer to consider this when trying to attach the link.
      receiver.setDesiredCapabilities(new Symbol[] { AmqpSupport.SHARED, AmqpSupport.GLOBAL });
    }

    return receiver;
}
 

@Override
public Symbol readValue()
{
    DecoderImpl decoder = getDecoder();
    int size = ((int)decoder.readRawByte()) & 0xff;
    return decoder.readRaw(_symbolCreator, size);
}
 

private Map<Symbol, Object> getConnetionProperties() {
    Map<Symbol, Object> properties = new HashMap<Symbol, Object>();

    properties.put(PRODUCT, "Qpid-JMS-NettyServer");
    properties.put(VERSION, "1.0");
    properties.put(PLATFORM, "java");

    return properties;
}
 

private void doSharedSubscriptionLinkCapabilitySupportedTestImpl(boolean durable) throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        // DONT include server connection capability to indicate support for shared-subs.
        // This will cause the link capability to be desired, and we verify success if offered.
        Symbol[] serverCapabilities = new Symbol[]{};

        Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities);
        connection.start();

        testPeer.expectBegin();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        String topicName = "myTopic";
        Topic dest = session.createTopic(topicName);
        String subscriptionName = "mySubscription";

        // Expect a shared receiver to attach, and succeed due to the server offering
        // the shared subs capability, i.e sharing is supported.
        if (durable) {
            Matcher<?> durableLinkNameMatcher = equalTo(subscriptionName);
            testPeer.expectSharedSubscriberAttach(topicName, subscriptionName, durableLinkNameMatcher, true, false, true, true, true);
            testPeer.expectLinkFlow();

            session.createSharedDurableConsumer(dest, subscriptionName);
        } else {
            Matcher<?> volatileLinkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "volatile1");
            testPeer.expectSharedSubscriberAttach(topicName, subscriptionName, volatileLinkNameMatcher, false, false, true, true, true);
            testPeer.expectLinkFlow();

            session.createSharedConsumer(dest, subscriptionName);
        }

        testPeer.expectClose();
        connection.close();

        testPeer.waitForAllHandlersToComplete(1000);
    }
}
 

@Override
public void setContentEncoding(String contentEncoding)
{
    if(_properties == null)
    {
        if(contentEncoding == null)
        {
            return;
        }
        _properties = new Properties();
    }
    _properties.setContentEncoding(Symbol.valueOf(contentEncoding));
}
 

@Test
public void testGetMessageAnnotations() {
   MessageImpl protonMessage = createProtonMessage();
   AMQPStandardMessage message = new AMQPStandardMessage(0, encodeMessage(protonMessage), null, null);

   MessageAnnotations decoded = message.getMessageAnnotations();
   assertNotSame(decoded, protonMessage.getMessageAnnotations());
   assertMessageAnnotationsEquals(protonMessage.getMessageAnnotations(), decoded);

   // Update the values
   decoded.getValue().put(Symbol.valueOf(UUID.randomUUID().toString()), "test");

   // Check that the message is unaffected.
   assertMessageAnnotationsNotEquals(protonMessage.getMessageAnnotations(), decoded);
}
 

protected FrameWithPayloadMatchingHandler(FrameType frameType,
                                            int channel,
                                            UnsignedLong numericDescriptor,
                                            Symbol symbolicDescriptor)
{
    super(frameType, channel, 0, numericDescriptor, symbolicDescriptor);
}
 

private void doMechanismSelectedExternalTestImpl(boolean requireClientCert, Symbol clientSelectedMech, Symbol[] serverMechs) throws Exception {
    TransportOptions sslOptions = new TransportOptions();
    sslOptions.setKeyStoreLocation(BROKER_JKS_KEYSTORE);
    sslOptions.setKeyStorePassword(PASSWORD);
    sslOptions.setVerifyHost(false);
    if (requireClientCert) {
        sslOptions.setTrustStoreLocation(BROKER_JKS_TRUSTSTORE);
        sslOptions.setTrustStorePassword(PASSWORD);
    }

    SSLContext context = TransportSupport.createJdkSslContext(sslOptions);

    try (TestAmqpPeer testPeer = new TestAmqpPeer(context, requireClientCert);) {
        String connOptions = "?transport.trustStoreLocation=" + CLIENT_JKS_TRUSTSTORE + "&" +
                             "transport.trustStorePassword=" + PASSWORD + "&" +
                             "jms.clientID=myclientid";
        if (requireClientCert) {
            connOptions += "&transport.keyStoreLocation=" + CLIENT_JKS_KEYSTORE + "&" +
                           "transport.keyStorePassword=" + PASSWORD;
        }

        testPeer.expectSaslFailingAuthentication(serverMechs, clientSelectedMech);

        JmsConnectionFactory factory = new JmsConnectionFactory("amqps://localhost:" + testPeer.getServerPort() + connOptions);
        try {
            factory.createConnection();
            fail("Expected exception to be thrown");
        } catch (JMSException jmse) {
            // Expected
        }

        testPeer.waitForAllHandlersToComplete(1000);
    }
}
 

@Test(timeout = 60000)
public void testClientIdIsSetInSubscriptionList() throws Exception {
   server.addAddressInfo(new AddressInfo(SimpleString.toSimpleString("mytopic"), RoutingType.ANYCAST));

   AmqpClient client = createAmqpClient();
   AmqpConnection connection = addConnection(client.connect());
   connection.setContainerId("testClient");
   connection.connect();

   try {
      AmqpSession session = connection.createSession();

      Source source = new Source();
      source.setDurable(TerminusDurability.UNSETTLED_STATE);
      source.setCapabilities(Symbol.getSymbol("topic"));
      source.setAddress("mytopic");
      session.createReceiver(source, "testSub");

      SimpleString fo = new SimpleString("testClient.testSub:mytopic");
      assertNotNull(server.locateQueue(fo));

   } catch (Exception e) {
      e.printStackTrace();
   } finally {
      connection.close();
   }
}
 

@Test
public void testCopy() {
    Map<Symbol, Object> properties = new HashMap<>();
    properties.put(Symbol.valueOf("x-opt"), "value");

    Open open = new Open();

    open.setContainerId("test");
    open.setHostname("host");
    open.setMaxFrameSize(UnsignedInteger.valueOf(42));
    open.setChannelMax(UnsignedShort.MAX_VALUE);
    open.setIdleTimeOut(UnsignedInteger.valueOf(111));
    open.setOfferedCapabilities(new Symbol[] { Symbol.valueOf("anonymous-relay") });
    open.setDesiredCapabilities(new Symbol[0]);
    open.setProperties(properties);

    Open copyOf = open.copy();

    assertEquals(open.getContainerId(), copyOf.getContainerId());
    assertEquals(open.getHostname(), copyOf.getHostname());
    assertEquals(open.getMaxFrameSize(), copyOf.getMaxFrameSize());
    assertEquals(open.getChannelMax(), copyOf.getChannelMax());
    assertEquals(open.getIdleTimeOut(), copyOf.getIdleTimeOut());
    assertArrayEquals(open.getDesiredCapabilities(), copyOf.getDesiredCapabilities());
    assertArrayEquals(open.getOfferedCapabilities(), copyOf.getOfferedCapabilities());
    assertEquals(open.getProperties(), copyOf.getProperties());
}
 

public AmqpRedirect(Map<Symbol, Object> redirect, AmqpProvider provider) {
    this.redirect = redirect;
    this.provider = provider;

    if (provider == null) {
        throw new IllegalArgumentException("A provider instance is required");
    }

    URI remoteURI = provider.getRemoteURI();
    if (remoteURI == null || remoteURI.getScheme() == null || remoteURI.getScheme().isEmpty()) {
        throw new IllegalArgumentException("The provider instance must provide a valid scheme");
    }
}
 

@Override
public void writeSymbol(final Symbol s)
{
    if(s == null)
    {
        writeNull();
    }
    else
    {
        _symbolType.fastWrite(this, s);
    }
}
 

@Test(timeout = 20000)
public void testSendFailsWhenDelayedDeliveryIsNotSupported() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {

        // DO NOT add capability to indicate server support for DELAYED-DELIVERY

        Connection connection = testFixture.establishConnecton(testPeer);

        connection.start();

        testPeer.expectBegin();

        Matcher<Symbol[]> desiredCapabilitiesMatcher = arrayContaining(new Symbol[] { DELAYED_DELIVERY });
        Symbol[] offeredCapabilities = null;
        testPeer.expectSenderAttach(notNullValue(), notNullValue(), false, false, false, false, 0, 1, null, null, desiredCapabilitiesMatcher, offeredCapabilities);

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        String topicName = "myTopic";
        Topic dest = session.createTopic(topicName);

        MessageProducer producer = session.createProducer(dest);
        producer.setDeliveryDelay(5000);

        // Producer should fail to send when message has delivery delay since remote
        // did not report that it supports that option.
        Message message = session.createMessage();
        try {
            producer.send(message);
            fail("Send should fail");
        } catch (JMSException jmsEx) {
            LOG.debug("Caught expected error from failed send.");
        }

        testPeer.expectClose();
        connection.close();

        testPeer.waitForAllHandlersToComplete(1000);
    }
}
 

public TargetMatcherCore()
{
    super(UnsignedLong.valueOf(0x0000000000000029L),
          Symbol.valueOf("amqp:target:list"));
}
 

/**
 * Verifies that subscriber cleanup occurs when the session it is on is remotely closed.
 *
 * @throws Exception if an unexpected error is encountered
 */
@Test(timeout = 20000)
public void testRemotelyEndSessionWithDurableSharedConsumer() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        // Add server connection capability to indicate support for shared-subs
        Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS};

        // Establish connection
        Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities);

        final CountDownLatch sessionClosed = new CountDownLatch(1);
        ((JmsConnection) connection).addConnectionListener(new JmsDefaultConnectionListener() {
            @Override
            public void onSessionClosed(Session session, Throwable exception) {
                sessionClosed.countDown();
            }
        });

        // Create first session
        testPeer.expectBegin();
        Session session1 = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        String topicName = "myTopic";
        Topic dest = session1.createTopic(topicName);

        String subscriptionName = "mySubscription";

        // Attach the first shared receiver on the first session
        Matcher<?> durableLinkNameMatcher = equalTo(subscriptionName);
        testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, durableLinkNameMatcher, true);
        testPeer.expectLinkFlow();

        MessageConsumer subscriber1 = session1.createSharedDurableConsumer(dest,  subscriptionName);

        // Create second session
        testPeer.expectBegin();
        Session session2 = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // Attach the second shared receiver on the second session
        durableLinkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "2");
        testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, durableLinkNameMatcher, true);

        testPeer.expectLinkFlow();

        // Then remotely end the session (and thus the subscriber along with it) after the flow is received.
        testPeer.remotelyEndLastOpenedSession(true, 0, AmqpError.RESOURCE_LIMIT_EXCEEDED, "TestingRemoteClosure");

        MessageConsumer subscriber2 = session2.createSharedDurableConsumer(dest,  subscriptionName);
        assertNotNull(subscriber2);

        assertTrue("Session closed callback didn't trigger", sessionClosed.await(5, TimeUnit.SECONDS));

        testPeer.waitForAllHandlersToComplete(1000);

        // Now try to unsubscribe (using first session, still open). It should fail due to sub still
        // being in use on the first session. No frames should be sent.
        try {
            session1.unsubscribe(subscriptionName);
            fail("Should have thrown a JMSException");
        } catch (JMSException ex) {
            // Expected
        }

        // Now close the first subscriber
        testPeer.expectDetach(false, true, false);
        subscriber1.close();

        testPeer.waitForAllHandlersToComplete(1000);

        // Try to unsubscribe again (using first session, still open), should now work.
        testPeer.expectDurableSubUnsubscribeNullSourceLookup(false, false, subscriptionName, topicName, true);
        testPeer.expectDetach(true, true, true);

        session1.unsubscribe(subscriptionName);

        testPeer.expectClose();
        connection.close();
    }
}
 

@Override
public Map<Symbol, Object> getRemoteProperties()
{
    return _remoteProperties;
}
 

@Override
public Map<Symbol, Object> getRemoteProperties() {
  return connection.getRemoteProperties();
}
 

public void expectDurableSubUnsubscribeNullSourceLookup(boolean failLookup, boolean shared, String subscriptionName, String topicName, boolean hasClientID) {
    String linkName = subscriptionName;
    if(!hasClientID) {
        linkName += AmqpSupport.SUB_NAME_DELIMITER + "global";
    }

    Matcher<String> linkNameMatcher = equalTo(linkName);
    Matcher<Object> nullSourceMatcher = nullValue();

    Source responseSourceOverride = null;
    Symbol errorType = null;
    String errorMessage = null;

    if(failLookup){
        errorType = AmqpError.NOT_FOUND;
        errorMessage = "No subscription link found";
    } else {
        responseSourceOverride = new Source();
        responseSourceOverride.setAddress(topicName);
        responseSourceOverride.setDynamic(false);
        //TODO: will possibly be changed to a 1/config durability
        responseSourceOverride.setDurable(TerminusDurability.UNSETTLED_STATE);
        responseSourceOverride.setExpiryPolicy(TerminusExpiryPolicy.NEVER);

        if(shared) {
            if(hasClientID) {
                responseSourceOverride.setCapabilities(new Symbol[]{SHARED});
            } else {
                responseSourceOverride.setCapabilities(new Symbol[]{SHARED, GLOBAL});
            }
        }
    }

    // If we don't have a ClientID, expect link capabilities to hint that we are trying
    // to reattach to a 'global' shared subscription.
    Matcher<?> linkDesiredCapabilitiesMatcher = null;
    if(!hasClientID) {
        linkDesiredCapabilitiesMatcher = arrayContaining(new Symbol[] { SHARED, GLOBAL });
    }

    expectReceiverAttach(linkNameMatcher, nullSourceMatcher, false, failLookup, false, false, errorType, errorMessage, responseSourceOverride, linkDesiredCapabilitiesMatcher, null);
}