import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
/**
 * Verifies that the endpoint rejects a client's attempt to create a link using <em>AT_MOST_ONCE</em>
 * delivery mode.
 */
@Test
public void testOnLinkAttachDisconnectsClientsUsingWrongQos() {

    ProtonConnection con = mock(ProtonConnection.class);
    ProtonReceiver receiver = mock(ProtonReceiver.class);
    when(receiver.getRemoteQoS()).thenReturn(ProtonQoS.AT_MOST_ONCE);
    ResourceIdentifier targetAddress = ResourceIdentifier.from("event", "tenant", null);

    endpoint.onLinkAttach(con, receiver, targetAddress);

    ArgumentCaptor<ErrorCondition> errorCondition = ArgumentCaptor.forClass(ErrorCondition.class);
    verify(receiver).setCondition(errorCondition.capture());
    assertThat(errorCondition.getValue(), is(ErrorConditions.ERROR_UNSUPPORTED_DELIVERY_MODE));
    verify(receiver).close();
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
/**
 * Verifies that all links to upstream clients are closed when the connection to the
 * downstream container is lost.
 */
@Test
public void testDownstreamDisconnectClosesUpstreamReceivers() {

    final UpstreamReceiver client = newClient();
    final ProtonSender downstreamSender = newMockSender(false);
    // expect the connection factory to be invoked twice
    // first on initial connection
    // second on re-connect attempt
    HandlerCapturingConnectionFactory factory = new HandlerCapturingConnectionFactory(con, 2);

    // GIVEN an adapter connected to a downstream container
    givenADownstreamAdapter(downstreamSender);
    adapter.setDownstreamConnectionFactory(factory);
    adapter.start(Future.future());
    adapter.addSender(client, downstreamSender);

    // WHEN the downstream connection fails
    factory.getDisconnectHandler().handle(con);

    // THEN the adapter tries to reconnect to the downstream container and has closed all upstream receivers
    factory.await(1, TimeUnit.SECONDS);
    verify(client).close(any(ErrorCondition.class));
    assertTrue(adapter.isActiveSendersEmpty());
    assertTrue(adapter.isSendersPerConnectionEmpty());
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
/**
 * Verifies that the endpoint does not open a link with a client that uses an unsupported
 * delivery mode.
 */
@Test
public void testOnLinkAttachClosesLinkIfClientWantsToUseUnsupportedDeliveryMode() {

    // GIVEN an endpoint
    MessageForwardingEndpoint<ServiceConfigProperties> endpoint = getEndpoint();

    // WHEN a client tries to attach using an unsupported delivery mode
    final ProtonConnection connection = mock(ProtonConnection.class);
    final ProtonReceiver receiver = mock(ProtonReceiver.class);
    final ResourceIdentifier targetAddress = ResourceIdentifier.fromString("telemetry/tenant");
    when(receiver.getRemoteQoS()).thenReturn(ProtonQoS.AT_LEAST_ONCE);
    endpoint.onLinkAttach(connection, receiver, targetAddress);

    // THEN the endpoint closes the link
    ArgumentCaptor<ErrorCondition> errorCondition = ArgumentCaptor.forClass(ErrorCondition.class);
    verify(receiver).setCondition(errorCondition.capture());
    assertThat(errorCondition.getValue(), is(ErrorConditions.ERROR_UNSUPPORTED_DELIVERY_MODE));
    verify(receiver).close();
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
/**
 * Verifies that a message containing a matching registration assertion is
 * forwarded to the downstream adapter.
 */
@Test
public void testForwardMessageAcceptsCorrectRegistrationAssertion() {

    final String validToken = getToken(SECRET, "tenant", "4711");
    UpstreamReceiver client = mock(UpstreamReceiver.class);
    ProtonDelivery delivery = mock(ProtonDelivery.class);
    DownstreamAdapter adapter = mock(DownstreamAdapter.class);
    when(tokenValidator.isValid(validToken, "tenant", "4711")).thenReturn(Boolean.TRUE);
    MessageForwardingEndpoint<ServiceConfigProperties> endpoint = getEndpoint();
    endpoint.setRegistrationAssertionValidator(tokenValidator);
    endpoint.setDownstreamAdapter(adapter);

    // WHEN processing a message bearing a valid registration assertion
    Message msg = ProtonHelper.message();
    MessageHelper.addRegistrationAssertion(msg, validToken);
    MessageHelper.addAnnotation(msg, MessageHelper.APP_PROPERTY_RESOURCE, "telemetry/tenant/4711");
    endpoint.forwardMessage(client, delivery, msg);

    // THEN the message is sent downstream
    verify(adapter).processMessage(client, delivery, msg);
    verify(client, times(0)).close(any(ErrorCondition.class));
    // and the assertion has been removed from the message
    assertThat("downstream message should not contain registration assertion",
            MessageHelper.getRegistrationAssertion(msg), is(nullValue()));
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
/**
 * Verifies that a message containing a registration assertion for a tenant
 * other than the one from the message's target address is rejected.
 */
@Test
public void testProcessMessageRejectsRegistrationAssertionForWrongTenant() {

    final String invalidToken = getToken(SECRET, "wrong-tenant", "4711");
    UpstreamReceiver client = mock(UpstreamReceiver.class);
    ProtonDelivery delivery = mock(ProtonDelivery.class);
    when(tokenValidator.isValid(invalidToken, "tenant", "4711")).thenReturn(Boolean.FALSE);
    MessageForwardingEndpoint<ServiceConfigProperties> endpoint = getEndpoint();
    endpoint.setRegistrationAssertionValidator(tokenValidator);

    Message msg = ProtonHelper.message();
    MessageHelper.addRegistrationAssertion(msg, invalidToken);
    MessageHelper.addAnnotation(msg, MessageHelper.APP_PROPERTY_RESOURCE, "telemetry/tenant/4711");
    endpoint.forwardMessage(client, delivery, msg);

    verify(delivery).disposition(any(Rejected.class), anyBoolean());
    verify(client, never()).close(any(ErrorCondition.class));
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
/** When happens when the partition filter < 0? */
@Test
public <K, V> void filters_negativeIntegerPartitionFilter() throws Exception {
	String topic = "my_topic";
	Vertx vertx = Vertx.vertx();
	AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, new AmqpBridgeConfigProperties());
	endpoint.open();
	ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, topic+"/group.id/blah");
	// Call handle()
	Map<Symbol, Object> filter = new HashMap<>();
	filter.put(Symbol.getSymbol(AmqpBridge.AMQP_PARTITION_FILTER), -1);
	filter.put(Symbol.getSymbol(AmqpBridge.AMQP_OFFSET_FILTER), 10L);
	((Source)mockSender.getRemoteSource()).setFilter(filter);
	endpoint.handle(new AmqpEndpoint(mockSender));
	
	ArgumentCaptor<ErrorCondition> errorCap = ArgumentCaptor.forClass(ErrorCondition.class);
	verify(mockSender).setCondition(errorCap.capture());
	verify(mockSender).close();
	
	assertDetach(mockSender, 
			AmqpBridge.AMQP_ERROR_WRONG_FILTER,
			"Wrong filter");
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
@Override
public void handleEnd(End end, Binary payload, Integer channel)
{
    TransportSession transportSession = _remoteSessions.get(channel);
    if(transportSession == null)
    {
        // TODO - fail due to attach on non-begun session
    }
    else
    {
        _remoteSessions.remove(channel);
        transportSession.receivedEnd();
        transportSession.unsetRemoteChannel();
        SessionImpl session = transportSession.getSession();
        session.setRemoteState(EndpointState.CLOSED);
        ErrorCondition errorCondition = end.getError();
        if(errorCondition != null)
        {
            session.getRemoteCondition().copyFrom(errorCondition);
        }

        _connectionEndpoint.put(Event.Type.SESSION_REMOTE_CLOSE, session);
    }
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
public ErrorCondition newInstance(Object described)
{
    List l = (List) described;

    ErrorCondition o = new ErrorCondition();

    if(l.isEmpty())
    {
        throw new DecodeException("The condition field cannot be omitted");
    }

    switch(3 - l.size())
    {

        case 0:
            o.setInfo( (Map) l.get( 2 ) );
        case 1:
            o.setDescription( (String) l.get( 1 ) );
        case 2:
            o.setCondition( (Symbol) l.get( 0 ) );
    }


    return o;
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
public Detach newInstance(Object described)
{
    List l = (List) described;

    Detach o = new Detach();

    if(l.isEmpty())
    {
        throw new DecodeException("The handle field cannot be omitted");
    }

    switch(3 - l.size())
    {

        case 0:
            o.setError( (ErrorCondition) l.get( 2 ) );
        case 1:
            Boolean closed = (Boolean) l.get(1);
            o.setClosed(closed == null ? false : closed);
        case 2:
            o.setHandle( (UnsignedInteger) l.get( 0 ) );
    }


    return o;
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
private void processEventSessionRemoteState(Event event) {
    final String methodName = "processEventSessionRemoteState";
    logger.entry(this, methodName, event);

    if (event.getSession().getRemoteState() == EndpointState.ACTIVE) {
        if (event.getSession().getLocalState() == EndpointState.ACTIVE) {
            final EngineConnection engineConnection =
                    (EngineConnection) event.getConnection().getContext();
            if (!engineConnection.closed) {
                // First session has opened on the connection
                OpenRequest req = engineConnection.openRequest;
                engineConnection.openRequest = null;
                engineConnection.requestor.tell(new OpenResponse(req, engineConnection), this);
            }
        } else {
            // The remote end is trying to establish a new session with us, which is not allowed. I don't think this is a usual case,
            // but could occur with a badly written remote end (i.e. sends an initial AMQP open immediately followed by a begin)
            final Connection protonConnection = event.getConnection();
            protonConnection.setCondition(new ErrorCondition(Symbol.getSymbol("mqlight:session-remote-open-rejected"),
                                                             "MQ Light client is unable to accept an open session request"));
            protonConnection.close();
        }
    }

    logger.exit(this, methodName);
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
@Override
public void onLinkRemoteOpen(Event e) {
    System.out.println("onLinkRemoteOpen");
    e.getLink().open();
    if (e.getLink() instanceof Sender) {
        Sender sender = (Sender)e.getLink();
        delivery = sender.delivery(new byte[]{1});
        sender.send(new byte[]{1, 2, 3}, 0, 3);
        sender.advance();
    } else {
        Receiver receiver = (Receiver)e.getLink();
        receiver.flow(1024);
    }
    if (closeConnection) {
        e.getConnection().setCondition(new ErrorCondition(Symbol.getSymbol("symbol"), "Something went wrong!"));
        e.getConnection().close();
    }
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
@Override
public void onResponse(IAmqpProtocolConverter converter, Response response) throws IOException {
    if (response.isException()) {
        sender.setSource(null);
        Throwable exception = ((ExceptionResponse) response).getException();
        Symbol condition = AmqpError.INTERNAL_ERROR;
        if (exception instanceof InvalidSelectorException) {
            condition = AmqpError.INVALID_FIELD;
        }
        sender.setCondition(new ErrorCondition(condition, exception.getMessage()));
        subscriptionsByConsumerId.remove(id);
        sender.close();
    } else {
        sessionContext.consumers.put(id, consumerContext);
        sender.open();
    }
    pumpProtonToSocket();
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
/**
 * Attempt to read and return the embedded error message in the given ErrorCondition
 * object.  If no message can be extracted a generic message is returned.
 *
 * @param errorCondition The ErrorCondition to extract the error message from.
 * @return an error message extracted from the given ErrorCondition.
 */
public static String extractErrorMessage(ErrorCondition errorCondition) {
   String message = "Received error from remote peer without description";
   if (errorCondition != null) {
      if (errorCondition.getDescription() != null && !errorCondition.getDescription().isEmpty()) {
         message = errorCondition.getDescription();
      }

      Symbol condition = errorCondition.getCondition();
      if (condition != null) {
         message = message + " [condition = " + condition + "]";
      }
   }

   return message;
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
@Override
public int sendMessage(MessageReference ref, Message message, ServerConsumer consumer, int deliveryCount) {

   ProtonServerSenderContext plugSender = (ProtonServerSenderContext) consumer.getProtocolContext();

   try {
      return plugSender.deliverMessage(ref, deliveryCount, transportConnection);
   } catch (Exception e) {
      connection.lock();
      try {
         plugSender.getSender().setCondition(new ErrorCondition(AmqpError.INTERNAL_ERROR, e.getMessage()));
         connection.flush();
      } finally {
         connection.unlock();
      }
      throw new IllegalStateException("Can't deliver message " + e, e);
   }

}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
@Override
public void close(ErrorCondition condition) throws ActiveMQAMQPException {
   closed = true;
   if (condition != null) {
      sender.setCondition(condition);
   }
   protonSession.removeSender(sender);
   connection.lock();
   try {
      sender.close();
   } finally {
      connection.unlock();
   }
   connection.flush();

   try {
      sessionSPI.closeSender(brokerConsumer);
   } catch (Exception e) {
      log.warn(e.getMessage(), e);
      throw new ActiveMQAMQPInternalErrorException(e.getMessage());
   }
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
public long tick(boolean firstTick) {
   lock.lock();
   try {
      if (!firstTick) {
         try {
            if (connection.getLocalState() != EndpointState.CLOSED) {
               long rescheduleAt = transport.tick(TimeUnit.NANOSECONDS.toMillis(System.nanoTime()));
               if (transport.isClosed()) {
                  throw new IllegalStateException("Channel was inactive for to long");
               }
               return rescheduleAt;
            }
         } catch (Exception e) {
            log.warn(e.getMessage(), e);
            transport.close();
            connection.setCondition(new ErrorCondition());
         }
         return 0;
      }
      return transport.tick(TimeUnit.NANOSECONDS.toMillis(System.nanoTime()));
   } finally {
      lock.unlock();
      flushBytes();
   }
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
public void addReceiver(Receiver receiver) throws Exception {
   try {
      ProtonServerReceiverContext protonReceiver = new ProtonServerReceiverContext(sessionSPI, connection, this, receiver);
      protonReceiver.initialise();
      receivers.put(receiver, protonReceiver);
      ServerProducer serverProducer = new ServerProducerImpl(receiver.getName(), "AMQP", receiver.getTarget().getAddress());
      sessionSPI.addProducer(serverProducer);
      receiver.setContext(protonReceiver);
      connection.lock();
      try {
         receiver.open();
      } finally {
         connection.unlock();
      }
   } catch (ActiveMQAMQPException e) {
      receivers.remove(receiver);
      receiver.setTarget(null);
      receiver.setCondition(new ErrorCondition(e.getAmqpError(), e.getMessage()));
      connection.lock();
      try {
         receiver.close();
      } finally {
         connection.unlock();
      }
   }
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
/**
 * Attempt to read and return the embedded error message in the given ErrorCondition
 * object.  If no message can be extracted a generic message is returned.
 *
 * @param errorCondition
 *      The ErrorCondition to extract the error message from.
 *
 * @return an error message extracted from the given ErrorCondition.
 */
public static String extractErrorMessage(ErrorCondition errorCondition) {
    String message = "Received error from remote peer without description";
    if (errorCondition != null) {
        if (errorCondition.getDescription() != null && !errorCondition.getDescription().isEmpty()) {
            message = errorCondition.getDescription();
        }

        Symbol condition = errorCondition.getCondition();
        if (condition != null) {
            message = message + " [condition = " + condition + "]";
        }
    }

    return message;
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
/**
 * When a redirect type exception is received this method is called to create the
 * appropriate redirect exception type containing the error details needed.
 *
 * @param provider
 * 		  the AMQP provider instance that originates this exception
 * @param error
 *        the Symbol that defines the redirection error type.
 * @param message
 *        the basic error message that should used or amended for the returned exception.
 * @param condition
 *        the ErrorCondition that describes the redirection.
 *
 * @return an Exception that captures the details of the redirection error.
 */
public static Exception createRedirectException(AmqpProvider provider, Symbol error, String message, ErrorCondition condition) {
    Exception result = null;
    Map<?, ?> info = condition.getInfo();

    if (info == null) {
        result = new IOException(message + " : Redirection information not set.");
    } else {
        @SuppressWarnings("unchecked")
        AmqpRedirect redirect = new AmqpRedirect((Map<Symbol, Object>) info, provider);

        try {
            result = new ProviderRedirectedException(message, redirect.validate().toURI());
        } catch (Exception ex) {
            result = new IOException(message + " : " + ex.getMessage());
        }
    }

    return result;
}
 

import org.apache.qpid.proton.amqp.transport.ErrorCondition; //导入依赖的package包/类
@Test
public void testCreateRedirectionExceptionWithNoNetworkHost() throws URISyntaxException {
    AmqpProvider mockProvider = Mockito.mock(AmqpProvider.class);
    Mockito.when(mockProvider.getRemoteURI()).thenReturn(new URI("amqp://localhost:5672"));

    ErrorCondition condition = new ErrorCondition();

    Map<Symbol, Object> info = new HashMap<>();
    info.put(AmqpSupport.PORT, "5672");
    info.put(AmqpSupport.OPEN_HOSTNAME, "localhost");
    info.put(AmqpSupport.SCHEME, "amqp");
    info.put(AmqpSupport.PATH, "websocket");

    condition.setInfo(info);

    Symbol error = AmqpError.INTERNAL_ERROR;
    String message = "Failed to connect";

    Exception result = AmqpSupport.createRedirectException(mockProvider, error, message, condition);

    assertNotNull(result);
    assertFalse(result instanceof ProviderRedirectedException);
    assertTrue(result instanceof IOException);
}