import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
@RabbitListener(queues = Constants.Queues.FILE_STAGED)
public void handleFileStagedEvent(FileMessage fileMessage) {
    if(!StringUtils.isEmpty(fileMessage.getContentType()) &&
        fileMessage.getMediaType().equals(FileMediaTypes.HCA_DATA_FILE)){
        try {
            fileService.updateStagedFileUrl(fileMessage.getStagingAreaId(),
                    fileMessage.getFileName(),
                    fileMessage.getCloudUrl());
        } catch (CoreEntityNotFoundException | RuntimeException e) {
            throw new AmqpRejectAndDontRequeueException(e.getMessage());
        }
    }
}
 

import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
/**
 * Method to validate if content type is set in the message properties.
 *
 * @param message
 *            the message to get validated
 */
private void checkContentTypeJson(final Message message) {
    if (message.getBody().length == 0) {
        return;
    }
    final MessageProperties messageProperties = message.getMessageProperties();
    final String headerContentType = (String) messageProperties.getHeaders().get("content-type");
    if (null != headerContentType) {
        messageProperties.setContentType(headerContentType);
    }
    final String contentType = messageProperties.getContentType();
    if (contentType != null && contentType.contains("json")) {
        return;
    }
    throw new AmqpRejectAndDontRequeueException("Content-Type is not JSON compatible");
}
 

import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
/**
 * Executed on an authentication request.
 * 
 * @param message
 *            the amqp message
 * @return the rpc message back to supplier.
 */
@RabbitListener(queues = "${hawkbit.dmf.rabbitmq.authenticationReceiverQueue:authentication_receiver}", containerFactory = "listenerContainerFactory")
public Message onAuthenticationRequest(final Message message) {
    checkContentTypeJson(message);
    final SecurityContext oldContext = SecurityContextHolder.getContext();
    try {
        return handleAuthenticationMessage(message);
    } catch (final RuntimeException ex) {
        throw new AmqpRejectAndDontRequeueException(ex);
    } finally {
        SecurityContextHolder.setContext(oldContext);
    }
}
 

import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
@Test
@Description("Tests the update of an action of a target without a exist action id")
public void updateActionStatusWithoutActionId() {
    when(controllerManagementMock.findActionWithDetails(any())).thenReturn(Optional.empty());
    final MessageProperties messageProperties = createMessageProperties(MessageType.EVENT);
    messageProperties.setHeader(MessageHeaderKey.TOPIC, EventTopic.UPDATE_ACTION_STATUS.name());
    final DmfActionUpdateStatus actionUpdateStatus = new DmfActionUpdateStatus(1L, DmfActionStatus.DOWNLOAD);
    final Message message = amqpMessageHandlerService.getMessageConverter().toMessage(actionUpdateStatus,
            messageProperties);

    try {
        amqpMessageHandlerService.onMessage(message, MessageType.EVENT.name(), TENANT, "vHost");
        fail("AmqpRejectAndDontRequeueException was excepeted since no action id was set");
    } catch (final AmqpRejectAndDontRequeueException exception) {
        // test ok - exception was excepted
    }
}
 

import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
@Test
@Description("Tests the update of an action of a target without a exist action id")
public void updateActionStatusWithoutExistActionId() {
    final MessageProperties messageProperties = createMessageProperties(MessageType.EVENT);
    messageProperties.setHeader(MessageHeaderKey.TOPIC, EventTopic.UPDATE_ACTION_STATUS.name());
    when(controllerManagementMock.findActionWithDetails(any())).thenReturn(Optional.empty());

    final DmfActionUpdateStatus actionUpdateStatus = createActionUpdateStatus(DmfActionStatus.DOWNLOAD);
    final Message message = amqpMessageHandlerService.getMessageConverter().toMessage(actionUpdateStatus,
            messageProperties);

    try {
        amqpMessageHandlerService.onMessage(message, MessageType.EVENT.name(), TENANT, "vHost");
        fail("AmqpRejectAndDontRequeueException was excepeted since no action id was set");
    } catch (final AmqpRejectAndDontRequeueException exception) {
        // test ok - exception was excepted
    }

}
 

import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
@StreamListener(UserLogBinding.CHANNEL_NAME)
void consume(@Validated UserLogMessage message) {
    try {
        if (logRepository.getByLogId(message.getLogId()) == null) {
            Log log = new Log();
            BeanUtils.copyProperties(message, log);
            logRepository.create(log);
        }
    } catch (Exception e) {
        throw new AmqpRejectAndDontRequeueException(e);
    }
}
 

import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
protected static void checkContentTypeJson(final Message message) {
    final MessageProperties messageProperties = message.getMessageProperties();
    if (messageProperties.getContentType() != null && messageProperties.getContentType().contains("json")) {
        return;
    }
    throw new AmqpRejectAndDontRequeueException("Content-Type is not JSON compatible");
}
 

import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
/**
 * * Executed if a amqp message arrives.
 * 
 * @param message
 *            the message
 * @param type
 *            the type
 * @param tenant
 *            the tenant
 * @param virtualHost
 *            the virtual host
 * @return the rpc message back to supplier.
 */
public Message onMessage(final Message message, final String type, final String tenant, final String virtualHost) {

    final SecurityContext oldContext = SecurityContextHolder.getContext();
    try {
        final MessageType messageType = MessageType.valueOf(type);
        switch (messageType) {
        case THING_CREATED:
            checkContentTypeJson(message);
            setTenantSecurityContext(tenant);
            registerTarget(message, virtualHost);
            break;
        case EVENT:
            checkContentTypeJson(message);
            setTenantSecurityContext(tenant);
            handleIncomingEvent(message);
            break;
        case PING:
            if (isCorrelationIdNotEmpty(message)) {
                amqpMessageDispatcherService.sendPingReponseToDmfReceiver(message, tenant, virtualHost);
            }
            break;
        default:
            logAndThrowMessageError(message, "No handle method was found for the given message type.");
        }
    } catch (final IllegalArgumentException ex) {
        throw new AmqpRejectAndDontRequeueException("Invalid message!", ex);
    } finally {
        SecurityContextHolder.setContext(oldContext);
    }
    return null;
}
 

import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
@Test
@Description("Tests not allowed content-type in message")
public void wrongContentType() {
    final MessageProperties messageProperties = new MessageProperties();
    messageProperties.setContentType("xml");
    final Message message = new Message(new byte[0], messageProperties);
    try {
        amqpMessageHandlerService.onMessage(message, MessageType.THING_CREATED.name(), TENANT, "vHost");
        fail("AmqpRejectAndDontRequeueException was excepeted due to worng content type");
    } catch (final AmqpRejectAndDontRequeueException e) {
    }
}
 

import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
@Test
@Description("Tests the creation of a thing without a 'reply to' header in message.")
public void createThingWitoutReplyTo() {
    final MessageProperties messageProperties = createMessageProperties(MessageType.THING_CREATED, null);
    messageProperties.setHeader(MessageHeaderKey.THING_ID, "1");
    final Message message = messageConverter.toMessage("", messageProperties);

    try {
        amqpMessageHandlerService.onMessage(message, MessageType.THING_CREATED.name(), TENANT, "vHost");
        fail("AmqpRejectAndDontRequeueException was excepeted since no replyTo header was set");
    } catch (final AmqpRejectAndDontRequeueException exception) {
        // test ok - exception was excepted
    }

}
 

import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
@Test
@Description("Tests the creation of a target/thing without a thingID by calling the same method that incoming RabbitMQ messages would access.")
public void createThingWithoutID() {
    final MessageProperties messageProperties = createMessageProperties(MessageType.THING_CREATED);
    final Message message = messageConverter.toMessage(new byte[0], messageProperties);
    try {
        amqpMessageHandlerService.onMessage(message, MessageType.THING_CREATED.name(), TENANT, "vHost");
        fail("AmqpRejectAndDontRequeueException was excepeted since no thingID was set");
    } catch (final AmqpRejectAndDontRequeueException exception) {
        // test ok - exception was excepted
    }
}
 

import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
@Test
@Description("Tests the call of the same method that incoming RabbitMQ messages would access with an unknown message type.")
public void unknownMessageType() {
    final String type = "bumlux";
    final MessageProperties messageProperties = createMessageProperties(MessageType.THING_CREATED);
    messageProperties.setHeader(MessageHeaderKey.THING_ID, "");
    final Message message = messageConverter.toMessage(new byte[0], messageProperties);

    try {
        amqpMessageHandlerService.onMessage(message, type, TENANT, "vHost");
        fail("AmqpRejectAndDontRequeueException was excepeted due to unknown message type");
    } catch (final AmqpRejectAndDontRequeueException exception) {
        // test ok - exception was excepted
    }
}
 

import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
protected static final void logAndThrowMessageError(final Message message, final String error) {
    LOGGER.warn("Warning! \"{}\" reported by message: {}", error, message);
    throw new AmqpRejectAndDontRequeueException(error);
}
 

import org.springframework.amqp.AmqpRejectAndDontRequeueException; //导入依赖的package包/类
public void handleMessage(RetryTestBean message) {
    unretry.countDown();
    throw new AmqpRejectAndDontRequeueException("test");
}