code: public Message onMessage(final Message message, final String type, final String tenant, final String virtualHost) {
          if (StringUtils.isEmpty(type) || StringUtils.isEmpty(tenant)) {
              throw new AmqpRejectAndDontRequeueException("Invalid message! tenant and type header are mandatory!");
          }
  
          final SecurityContext oldContext = SecurityContextHolder.getContext();
          try {
              final MessageType messageType = MessageType.valueOf(type);
              switch (messageType) {
              case THING_CREATED:
                  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;
      }
docs: * 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.