Java 类com.rabbitmq.client.AMQP.BasicProperties 实例源码

public static void main(String[] args) throws IOException {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setUsername("guest");
    factory.setPassword("guest");
    factory.setVirtualHost("/");
    factory.setHost("localhost");
    factory.setPort(5672);
    Connection newConnection = factory.newConnection();

    Channel channel = newConnection.createChannel();

    Scanner scanner = new Scanner(System.in);
    String message = "";
    while(!message.equals("exit")){
        System.out.println("Enter your message");
        message = scanner.next();
        channel.queueDeclare("flink-test", true, false, false, null);
        channel.basicPublish("", "flink-test", new BasicProperties.Builder()
                .correlationId(java.util.UUID.randomUUID().toString()).build(), message.getBytes());
    }

    scanner.close();
    channel.close();
    newConnection.close();
}

public static void main(String[] args) throws IOException {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setUsername("guest");
    factory.setPassword("guest");
    factory.setVirtualHost("/");
    factory.setHost("localhost");
    factory.setPort(5672);
    Connection newConnection = factory.newConnection();

    Channel channel = newConnection.createChannel();

    Scanner scanner = new Scanner(System.in);
    String message = "";
    while(!message.equals("exit")){
        System.out.println("Enter your message");
        message = scanner.next();
        channel.queueDeclare("flink-test", true, false, false, null);
        channel.basicPublish("", "flink-test", new BasicProperties.Builder()
                .correlationId(java.util.UUID.randomUUID().toString()).build(), message.getBytes());
    }

    scanner.close();
    channel.close();
    newConnection.close();
}

/**
 * @param args
 * @throws TimeoutException
 * @throws IOException
 * @throws InterruptedException
 * @throws ConsumerCancelledException
 * @throws ShutdownSignalException
 */
public static void main(String[] args) throws IOException, TimeoutException, ShutdownSignalException,
        ConsumerCancelledException, InterruptedException {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost(host);
    factory.setUsername(username);
    factory.setPassword(password);
    factory.setPort(port);
    factory.setVirtualHost(virtualHost);
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();
    channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
    channel.basicQos(1);
    QueueingConsumer consumer = new QueueingConsumer(channel);
    channel.basicConsume(RPC_QUEUE_NAME, false, consumer);

    System.out.println("RPCServer Awating RPC request");
    while (true) {
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        BasicProperties props = delivery.getProperties();
        BasicProperties replyProps = new AMQP.BasicProperties.Builder().correlationId(props.getCorrelationId())
                .build();

        String message = new String(delivery.getBody(), "UTF-8");
        int n = Integer.parseInt(message);

        System.out.println("RPCServer fib(" + message + ")");
        String response = "" + fib(n);
        channel.basicPublish("", props.getReplyTo(), replyProps, response.getBytes());
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
}

@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
    if(body != null) {
        try {
            // resolve the returned message
            long deliveryTag = envelope.getDeliveryTag();
            String message = new String(body, "UTF-8");
            LOG.info("Task received: consumerTag=" + consumerTag + ", deliveryTag=" + deliveryTag + ", message=" + message);

            JSONObject taskReq = JSONObject.parseObject(message);
            if(!taskReq.isEmpty()) {
                long id = messageIdGenerator.incrementAndGet();
                WaitingTask task = new WaitingTask(id, taskReq, getChannel(), deliveryTag);
                waitingTasks.putLast(task);
                LOG.info("Add task to waiting queue: " + task);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

/****
 * This method is used to publish the message to RabbitMQ
 * @param routingKey
 * @param msg is Eiffel Event
 * @throws IOException
 */
public void send(String routingKey, String msg) throws IOException {

    Channel channel = giveMeRandomChannel();
    channel.addShutdownListener(new ShutdownListener() {
        public void shutdownCompleted(ShutdownSignalException cause) {
            // Beware that proper synchronization is needed here
            if (cause.isInitiatedByApplication()) {
                log.debug("Shutdown is initiated by application. Ignoring it.");
            } else {
                log.error("Shutdown is NOT initiated by application.");
                log.error(cause.getMessage());
                boolean cliMode = Boolean.getBoolean(PropertiesConfig.CLI_MODE);
                if (cliMode) {
                    System.exit(-3);
                }
            }
        }
    });

    BasicProperties msgProps = MessageProperties.BASIC;
    if (usePersitance)
        msgProps = MessageProperties.PERSISTENT_BASIC;

    channel.basicPublish(exchangeName, routingKey, msgProps, msg.getBytes());
    log.info("Published message with size {} bytes on exchange '{}' with routing key '{}'", msg.getBytes().length,
            exchangeName, routingKey);
}

public String call(String message) throws Exception {
    String response = null;
    String corrId = UUID.randomUUID().toString();

    BasicProperties props = new BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();

    channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));

    while (true) {
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        if (delivery.getProperties().getCorrelationId().equals(corrId)) {
            response = new String(delivery.getBody(), "UTF-8");
            break;
        }
    }

    return response;
}

/**
 * Sends the given command to the command queue with the given data appended
 * and using the given properties.
 *
 * @param address
 *            address for the message
 * @param command
 *            the command that should be sent
 * @param data
 *            data that should be appended to the command
 * @param props
 *            properties that should be used for the message
 * @throws IOException
 */
protected void sendToCmdQueue(String address, byte command, byte data[], BasicProperties props) throws IOException {
    byte sessionIdBytes[] = RabbitMQUtils.writeString(address);
    // + 5 because 4 bytes for the session ID length and 1 byte for the
    // command
    int dataLength = sessionIdBytes.length + 5;
    boolean attachData = (data != null) && (data.length > 0);
    if (attachData) {
        dataLength += data.length;
    }
    ByteBuffer buffer = ByteBuffer.allocate(dataLength);
    buffer.putInt(sessionIdBytes.length);
    buffer.put(sessionIdBytes);
    buffer.put(command);
    if (attachData) {
        buffer.put(data);
    }
    cmdChannel.basicPublish(Constants.HOBBIT_COMMAND_EXCHANGE_NAME, "", props, buffer.array());
}

public String call(String message) throws Exception {
    String response;
    String corrId = UUID.randomUUID().toString();

    BasicProperties props = new BasicProperties
                                .Builder()
                                .correlationId(corrId)
                                .replyTo(replyQueueName)
                                .build();

    channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));

    while (true) {
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        if (delivery.getProperties().getCorrelationId().equals(corrId)) {
          response = new String(delivery.getBody(), "UTF-8");
          break;
        }
    }

    return response;
}

public String call(String message) throws Exception {
    String response = null;
    String corrId = UUID.randomUUID().toString();

    BasicProperties props = new BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();

    channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));

    while (true) {
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        if (delivery.getProperties().getCorrelationId().equals(corrId)) {
            response = new String(delivery.getBody(), "UTF-8");
            break;
        }
    }

    return response;
}

public byte[] call(String methodName, String message) throws Exception {
  byte[] response = null;
  String corrId = UUID.randomUUID().toString();

  BasicProperties props = new BasicProperties
      .Builder()
      .correlationId(corrId)
      .replyTo(replyQueueName)
      .build();

  channel.basicPublish("", REQUEST_QUEUE_NAME, props, message.getBytes("UTF-8"));

  while (true) {
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    if (delivery.getProperties().getCorrelationId().equals(corrId)) {
      response = delivery.getBody();
      break;
    }
  }

  return response;
}

private void consume(final String consumerTag, final Envelope envelope, final BasicProperties properties,
        final byte[] body) {
    final Message rawMessage = Message.builder()
            .body(body)
            .contentType(properties.getContentType())
            .encoding(properties.getContentEncoding())
            .type(properties.getType())
            .build();
    final T message = deserialize(rawMessage);
    final Consumer<Void> onAck = ackAction(envelope.getDeliveryTag());
    final MessageContext<T> messageContext = new RabbitMessageContext<>(
            message, consumerTag, envelope.getExchange(), envelope.getRoutingKey(),
            properties, onAck);
    logDelivery(messageContext);
    consumer.accept(messageContext);
}

@Test
public void testConnectionFactory() throws Exception {
    Assert.assertNotNull(connectionFactory1);
    Assert.assertNotNull(queue);

    RabbitmqConnection connection = connectionFactory1.getConnection();
    Assert.assertNotNull(connection);
    String queueName = "testing";
    Channel channel = connection.createChannel();
    channel.queueDeclare(queueName, false, false, false, null);
    String message = "Hello World!";

    final CountDownLatch counter = new CountDownLatch(1);
    Consumer consume = new DefaultConsumer(channel) {
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope,
                BasicProperties properties, byte[] body) throws IOException {
            Assert.assertEquals("Hello World!", new String(body));
            counter.countDown();
        }
    };

    channel.basicConsume(queueName, true, consume);
    channel.basicPublish("", queueName, null, message.getBytes());
    counter.await(10, TimeUnit.SECONDS);
    Assert.assertEquals(0, counter.getCount());
    channel.close();

}

public void setup() throws Exception {
    if (log.isTraceEnabled()) {
        log.trace("setup()");
    }
    this.consumer = new DefaultConsumer(this.channel) {

        @Override
        public void handleDelivery(String consumerTag, Envelope envelope,
                BasicProperties properties, byte[] body) throws IOException {
            MessageEndpoint localEndpoint;
            try {
                localEndpoint = endpointFactory.createEndpoint(null);
                RabbitmqBytesMessage m = new RabbitmqBytesMessage(consumerTag,envelope,properties,body);
                onMessage(localEndpoint, m);
            } catch (UnavailableException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

    };
    if ("javax.jms.Queue".equals(this.spec.getDestinationType())) {
        RabbitmqAdminQueueImpl queue = Util.lookup(new InitialContext(), this.spec.getDestination(), RabbitmqAdminQueueImpl.class);
        this.channel.basicConsume(queue.getDestinationAddress(),true, consumer);
    }
}

/**
 * Handles a message delivery from the broker.
 */
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
    BasicProperties properties, byte[] body) throws IOException {

    LOG.debug("Consumer: {} Received handle delivery", consumerTag);

    Message message = new Message(properties, body, envelope.getExchange(),
        envelope.getRoutingKey(), envelope.getDeliveryTag());

    LOG.info("Consumer: {} Received message: {}", consumerTag,
        envelope.getDeliveryTag());

    handleMessage(message);
    _channel.basicAck(envelope.getDeliveryTag(), true);
}

public static MessageProperties valueOf(BasicProperties bp) {
    MessageProperties mp = new MessageProperties();
    mp.setAppId(bp.getAppId());
    mp.setClusterId(bp.getClusterId());
    mp.setContentEncoding(bp.getContentEncoding());
    mp.setContentType(bp.getContentType());
    mp.setCorrelationId(bp.getCorrelationId());
    mp.setDeliveryMode(bp.getDeliveryMode());
    mp.setExpiration(bp.getExpiration());
    mp.setHeaders(bp.getHeaders());
    mp.setMessageId(bp.getMessageId());
    mp.setPriority(bp.getPriority());
    mp.setReplyTo(bp.getReplyTo());
    mp.setTimestamp(bp.getTimestamp());
    mp.setType(bp.getType());
    mp.setUserId(bp.getUserId());
    return mp;
}

private String rpcCall(String message) throws Exception {
    String QUEUE_NAME = "TEST_RPC";

    String replyQueueName = channel.queueDeclare().getQueue();
    System.out.println("replyQueueName: " + replyQueueName);

    QueueingConsumer consumer = new QueueingConsumer(channel);
    channel.basicConsume(replyQueueName, true, consumer);

    String response = null;
    String corrId = java.util.UUID.randomUUID().toString();

    BasicProperties props = new BasicProperties.Builder()
            .correlationId(corrId).replyTo(replyQueueName).build();

    channel.basicPublish("", QUEUE_NAME, props, message.getBytes());

    while (true) {
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        if (delivery.getProperties().getCorrelationId().equals(corrId)) {
            response = new String(delivery.getBody());
            break;
        }
    }
    return response;
}

public void setup() throws Exception {
    if (log.isTraceEnabled()) {
        log.trace("setup()");
    }
    this.consumer = new DefaultConsumer(this.channel) {

        @Override
        public void handleDelivery(String consumerTag, Envelope envelope,
                BasicProperties properties, byte[] body) throws IOException {
            MessageEndpoint localEndpoint;
            try {
                localEndpoint = endpointFactory.createEndpoint(null);
                RabbitmqBytesMessage m = new RabbitmqBytesMessage(consumerTag,envelope,properties,body);
                onMessage(localEndpoint, m);
            } catch (UnavailableException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

    };
    if ("javax.jms.Queue".equals(this.spec.getDestinationType())) {
        RabbitmqAdminQueueImpl queue = Util.lookup(new InitialContext(), this.spec.getDestination(), RabbitmqAdminQueueImpl.class);
        this.channel.basicConsume(queue.getDestinationAddress(),true, consumer);
    }
}

public void publish(EngineOutput o) {
  if(conn == null) connect();
  try {
    if(channel != null) {
      String rk = "logstream." + o.uuid;
      BasicProperties props = new BasicProperties();
      props.setContentType("applications/json");
      StringBuffer output = new StringBuffer();
      output.append("{\""+o.uuid+"\":[");
      if(o.object != null) {
        for(int i=0;i<o.object.length;i++) {
          if(i>0) output.append(",");
          output.append(o.object[i]);
        }
      }
      output.append("]}");
      channel.basicPublish(exchangeName, rk, props, output.toString().getBytes());
    }
  }
  catch(Exception e) {
    if(conn != null) conn.abort();
    conn = null;
    channel = null;
  }
}

@Override
public void handleDelivery( String consumerTag, Envelope envelope, BasicProperties properties, byte[] body )
throws IOException {

    try {
        Message message = SerializationUtils.deserializeObject( body );
        this.logger.finer( this.sourceName + " received a message " + message.getClass().getSimpleName()
                + " on routing key '" + envelope.getRoutingKey() + "'.");

        this.messageQueue.add( message );

    } catch( ClassNotFoundException | IOException e ) {
        this.logger.severe( this.sourceName + ": a message could not be deserialized. => " + e.getClass().getSimpleName());
        Utils.logException( this.logger, e );
        this.messageQueue.errorWhileReceivingMessage();
    }
}

/**
 * Unregisters an remote object previously registered into the bodies table
 *
 * @param urn
 *            the urn under which the active object has been registered
 */
public void unregister(URI uri) throws ProActiveException {
    String queueName = AMQPUtils.computeQueueNameFromURI(uri);

    ReusableChannel channel = null;
    try {
        channel = AMQPFederationUtils.getChannel(uri);
        channel.getChannel().basicPublish(
                AMQPFederationConfig.PA_AMQP_FEDERATION_RPC_EXCHANGE_NAME.getValue(),
                queueName,
                new BasicProperties.Builder().type(
                        AMQPFederationRemoteObjectServer.DELETE_QUEUE_MESSAGE_TYPE).build(), null);
        channel.returnChannel();
    } catch (IOException e) {
        if (channel != null) {
            channel.close();
        }
        throw new ProActiveException("Failed to delete object's queue", e);
    }
}

private void setProps () {
        this.PROPS = new BasicProperties
                .Builder()
                .contentEncoding(this.compressionCodec)
                .contentType(this.contentType)
//                .messageId(UUID.randomUUID().toString())
//                .timestamp(new Date())
                .deliveryMode(2)
                .priority(0)
                .build();
    }

@Override
public void emit(final Event event) {
    requireNonNull(event, "Cannot emit a null event!");

    final BasicProperties props = new BasicProperties().builder()
            .contentType("application/ld+json").contentEncoding("UTF-8").build();

    service.serialize(event).ifPresent(message -> {
        try {
            channel.basicPublish(exchangeName, queueName, mandatory, immediate, props, message.getBytes());
        } catch (final IOException ex) {
            LOGGER.error("Error writing to broker: {}", ex.getMessage());
        }
    });
}

@BeforeEach
public void setUp() throws IOException {
    initMocks(this);
    when(mockEvent.getTarget()).thenReturn(of(rdf.createIRI("trellis:repository/resource")));
    when(mockEvent.getAgents()).thenReturn(singleton(Trellis.AdministratorAgent));
    when(mockEvent.getCreated()).thenReturn(time);
    when(mockEvent.getIdentifier()).thenReturn(rdf.createIRI("urn:test"));
    when(mockEvent.getTypes()).thenReturn(singleton(AS.Update));
    when(mockEvent.getTargetTypes()).thenReturn(singleton(LDP.RDFSource));
    when(mockEvent.getInbox()).thenReturn(empty());
    doNothing().when(mockChannel).basicPublish(eq(exchangeName), eq(queueName), anyBoolean(), anyBoolean(),
            any(BasicProperties.class), any(byte[].class));
}

@Test
public void testAmqp() throws IOException {
    final EventService svc = new AmqpPublisher(mockChannel, exchangeName, queueName);
    svc.emit(mockEvent);

    verify(mockChannel).basicPublish(eq(exchangeName), eq(queueName), anyBoolean(), anyBoolean(),
            any(BasicProperties.class), any(byte[].class));
}

@Test
public void testError() throws IOException {
    doThrow(IOException.class).when(mockChannel).basicPublish(eq(exchangeName), eq(queueName),
            anyBoolean(), anyBoolean(), any(BasicProperties.class), any(byte[].class));

    final EventService svc = new AmqpPublisher(mockChannel, exchangeName, queueName, true, true);
    svc.emit(mockEvent);

    verify(mockChannel).basicPublish(eq(exchangeName), eq(queueName), anyBoolean(), anyBoolean(),
            any(BasicProperties.class), any(byte[].class));
}

@Override
public void preProcess(Consumer t, Object proxy, Method method, Object[] args) {

    Map<String, Object> params = new HashMap<String, Object>();
    params.put(CaptureConstants.INFO_CLIENT_REQUEST_URL, url);
    params.put(CaptureConstants.INFO_CLIENT_REQUEST_ACTION, "Consumer." + method.getName());
    params.put(CaptureConstants.INFO_CLIENT_APPID, applicationId);
    params.put(CaptureConstants.INFO_CLIENT_TYPE, "rabbitmq.client");

    if (logger.isDebugable()) {
        logger.debug("Invoke START:" + url + ",op=Consumer." + method.getName(), null);
    }

    UAVServer.instance().runMonitorCaptureOnServerCapPoint(CaptureConstants.CAPPOINT_APP_CLIENT,
            Monitor.CapturePhase.PRECAP, params);
    // 调用链只关心真正消费消息
    if (method.getName().equals("handleDelivery")) {

        AMQP.BasicProperties props = (BasicProperties) args[2];
        if (props.getHeaders() != null
                && props.getHeaders().containsKey(InvokeChainConstants.PARAM_MQHEAD_SPANINFO)) {
            params.put(InvokeChainConstants.PARAM_MQHEAD_SPANINFO,
                    props.getHeaders().get(InvokeChainConstants.PARAM_MQHEAD_SPANINFO) + "");
            params.put(CaptureConstants.INFO_APPSERVER_CONNECTOR_REQUEST_URL, url);
        }

        // register adapter
        UAVServer.instance().runSupporter("com.creditease.uav.apm.supporters.InvokeChainSupporter",
                "registerAdapter", RabbitmqConsumerAdapter.class);

        UAVServer.instance().runSupporter("com.creditease.uav.apm.supporters.InvokeChainSupporter", "runCap",
                InvokeChainConstants.CHAIN_APP_SERVICE, InvokeChainConstants.CapturePhase.PRECAP, params,
                RabbitmqConsumerAdapter.class, args);
    }

}

@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
        throws IOException {
    Packet p = mDecoder.decode(Unpooled.wrappedBuffer(body));
    if (p != null) {
        TopicPacket packet = (TopicPacket) p;
        mServer.sendToSubscribers(packet.topic, packet.message);
    }
}

@Override
public void sendResponse(byte[] data, String responseQueueName, String correlId) {
    try {
        BasicProperties props = (new BasicProperties.Builder()).correlationId(correlId).build();
        responseChannel.basicPublish("", responseQueueName, props, data);
    } catch (Exception e) {
        LOGGER.error("Exception while sending response.", e);
    }
}

@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
        throws IOException {
    Optional.ofNullable(body).ifPresent(b -> {
        try {
            // resolve the returned message
            long deliveryTag = envelope.getDeliveryTag();
            String message = new String(b);
            LOG.info("Message received: deliveryTag=" + deliveryTag + ", message=" + message);

            JSONObject heartbeatMessage = JSONObject.parseObject(message);
            if(heartbeatMessage.containsKey(JSONKeys.TYPE)) {
                String type = heartbeatMessage.getString(JSONKeys.TYPE);
                switch(type) {
                case ScheduledConstants.HEARTBEAT_TYPE_TASK_PROGRESS:
                    if(taskResponseHandlingController.isValid(heartbeatMessage)) {
                        Heartbeat hb = new Heartbeat(getChannel(), heartbeatMessage, deliveryTag);
                        rawHeartbeatMessages.add(hb);
                        LOG.info("Added to rawHeartbeatMessages: " + hb);
                    } else {
                        LOG.warn("Invalid tasks response: " + heartbeatMessage);
                        sendAck(getChannel(), deliveryTag);
                    }
                    break;
                default:
                    LOG.warn("Unknown heartbeat: type=" + type + ", heartbeat=" + heartbeatMessage);
                }
            } else {
                // Ack unknown message
                sendAck(getChannel(), deliveryTag);
            }
        } catch (Exception e) {
            LOG.warn("Fail to consume message: ", e);
        }
    });
}

@Override
public void run() throws Exception {
    LOGGER.info("Sending request...");
    BasicProperties props = new BasicProperties.Builder().deliveryMode(2)
            .replyTo(Constants.CONTROLLER_2_FRONT_END_QUEUE_NAME).build();
    frontEnd2Controller.basicPublish("", Constants.FRONT_END_2_CONTROLLER_QUEUE_NAME, props,
            new byte[] { FrontEndApiCommands.LIST_AVAILABLE_BENCHMARKS });
    LOGGER.info("Waiting for response...");
    QueueingConsumer.Delivery delivery = consumer.nextDelivery(REQUEST_TIMEOUT);
    if (delivery == null) {
        throw new IOException("Didn't got a response after \"" + REQUEST_TIMEOUT + "\" ms.");
    }
    // parse the response
    String jsonString = RabbitMQUtils.readString(delivery.getBody());
    Gson gson = new Gson();
    Collection<BenchmarkMetaData> benchmarks = gson.fromJson(jsonString,
            new TypeToken<Collection<BenchmarkMetaData>>() {
            }.getType());
    // print results
    StringBuilder builder = new StringBuilder();
    builder.append("Response:");
    builder.append(NEWLINE);
    builder.append("benchmarks:");
    for (BenchmarkMetaData benchmark : benchmarks) {
        builder.append(NEWLINE);
        builder.append("\tname: ");
        builder.append(benchmark.benchmarkName);
        builder.append(NEWLINE);
        builder.append("\turi: ");
        builder.append(benchmark.benchmarkUri);
        builder.append(NEWLINE);
        builder.append("\tdescription: ");
        builder.append(benchmark.benchmarkDescription);
        builder.append(NEWLINE);
    }
    LOGGER.info(builder.toString());
}

@Override
public void run() throws Exception {
    Map<String, String> env = System.getenv();
    String benchmarkUri = env.getOrDefault(BENCHMARK_URI_KEY, null);
    if (benchmarkUri == null) {
        LOGGER.error("Couldn't get value of " + BENCHMARK_URI_KEY + ". Aborting.");
        throw new Exception("Couldn't get value of " + BENCHMARK_URI_KEY + ". Aborting.");
    }
    String systemUri = env.getOrDefault(SYSTEM_URI_KEY, null);
    if (systemUri == null) {
        LOGGER.error("Couldn't get value of " + SYSTEM_URI_KEY + ". Aborting.");
        throw new Exception("Couldn't get value of " + SYSTEM_URI_KEY + ". Aborting.");
    }
    String benchmarkModelFile = env.getOrDefault(BENCHMARK_PARAM_FILE_KEY, null);
    if (benchmarkModelFile == null) {
        LOGGER.error("Couldn't get value of " + BENCHMARK_PARAM_FILE_KEY + ". Aborting.");
        throw new Exception("Couldn't get value of " + BENCHMARK_PARAM_FILE_KEY + ". Aborting.");
    }
    LOGGER.info("Reading model from " + benchmarkModelFile + ".");
    Model model = readModel(benchmarkModelFile);
    byte[] data = RabbitMQUtils.writeByteArrays(new byte[] { FrontEndApiCommands.ADD_EXPERIMENT_CONFIGURATION },
            new byte[][] { RabbitMQUtils.writeString(benchmarkUri), RabbitMQUtils.writeString(systemUri),
                    RabbitMQUtils.writeModel(model) }, null);

    LOGGER.info("Sending request...");
    BasicProperties props = new BasicProperties.Builder().deliveryMode(2)
            .replyTo(Constants.CONTROLLER_2_FRONT_END_QUEUE_NAME).build();
    frontEnd2Controller.basicPublish("", Constants.FRONT_END_2_CONTROLLER_QUEUE_NAME, props, data);
    LOGGER.info("Waiting for response...");
    QueueingConsumer.Delivery delivery = consumer.nextDelivery(REQUEST_TIMEOUT);
    if (delivery == null) {
        throw new IOException(
                "Didn't got a response after \"" + REQUEST_TIMEOUT + "\" ms.");
    }
    // parse the response
    LOGGER.info("Response: " + RabbitMQUtils.readString(delivery.getBody()));
}

private Topic(String name, String exchange, Boolean durable, boolean autoDelete, boolean exclusive,
    BasicProperties properties) {
  this.name = name;
  this.exchange = exchange;
  this.durable = durable;
  this.autoDelete = autoDelete;
  this.exclusive = exclusive;
  this.properties = properties;
}

@Override
public void handleDelivery(final String consumerTag, final Envelope envelope,
        final BasicProperties properties, final byte[] body)
                throws IOException {
    executor.execute(() -> {
        try {
            consume(consumerTag, envelope, properties, body);
        } catch (final Exception e) {
            tryNotifyConsumeError(e);
        }
    });
}

RabbitmqBytesMessage(String consumerTag, Envelope envelope,
        BasicProperties properties, byte[] body) {
    if (body == null)
        throw new IllegalArgumentException("Message body cannot be null");
    this.consumerTag = consumerTag;
    this.envelope = envelope;
    this.properties = properties;
    this.body = body;
    ByteArrayInputStream bais = new ByteArrayInputStream(this.body);
    dis = new DataInputStream(bais);
}

private void processDelivery(Delivery delivery) throws Exception {
    BasicProperties props = delivery.getProperties();
    BasicProperties replyProps = new BasicProperties.Builder().correlationId(props.getCorrelationId()).build();
    DataInputStream data = new DataInputStream(new InflaterInputStream(new ByteArrayInputStream(delivery.getBody())));
    RenderMode mode = RenderMode.values()[data.readUnsignedByte()];
    int width = data.readUnsignedShort();
    int height = data.readUnsignedShort();
    GameProfile profile = Profiles.readGameProfile(data);
    Map<String, String[]> params = Maps.newHashMap();
    int len = data.readUnsignedShort();
    for (int i = 0; i < len; i++) {
        String key = data.readUTF();
        String[] val = new String[data.readUnsignedByte()];
        for (int v = 0; v < val.length; v++) {
            val[v] = data.readUTF();
        }
        params.put(key, val);
    }
    byte[] skinData = new byte[data.readInt()];
    data.readFully(skinData);
    BufferedImage skin = new PngImage().read(new ByteArrayInputStream(skinData), false);
    Visage.log.info("Received a job to render a "+width+"x"+height+" "+mode.name().toLowerCase()+" for "+(profile == null ? "null" : profile.getName()));

    RenderConfiguration conf = new RenderConfiguration(Type.fromMode(mode), Profiles.isSlim(profile), mode.isTall(), Profiles.isFlipped(profile));

    glClearColor(0, 0, 0, 0);
    glClear(GL_COLOR_BUFFER_BIT | GL_DEPTH_BUFFER_BIT);
    byte[] pngBys = draw(conf, width, height, profile, skin, params);
    if (Visage.trace) Visage.log.finest("Got png bytes");
    parent.channel.basicPublish("", props.getReplyTo(), replyProps, buildResponse(0, pngBys));
    if (Visage.trace) Visage.log.finest("Published response");
    parent.channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    if (Visage.trace) Visage.log.finest("Ack'd message");
}

public void handleDelivery(final String consumerTag,
        final Envelope envelope, final BasicProperties props,
        final byte[] body) throws IOException {

    Event<?> event = (Event<?>) SerializationUtils.deserialize(body);
    internalDelegateHandling(event);

    // Acknowledge the received message after it has been handled
    this.endPoint.channel().basicAck(envelope.getDeliveryTag(), false);
}

@Override protected void publish(Event<?> event) throws IOException {
    byte[] data = SerializationUtils.serialize(event);
    BasicProperties props = new BasicProperties.Builder()
            .messageId(event.identifier().toString()).deliveryMode(2)
            .priority(0).type(event.getClass().getCanonicalName()).build();

    Channel c = endPoint.channel();
    c.basicPublish(this.exchange, this.routingKey, props, data);
}

@Test
public void rpcServer() throws Exception {
    String QUEUE_NAME = "TEST_RPC";
    //
    DeclareOk declare = channel.queueDeclare(QUEUE_NAME, false, false,
            false, null);
    // System.out.println("declare: " + declare);
    channel.basicQos(1);

    QueueingConsumer consumer = new QueueingConsumer(channel);
    channel.basicConsume(QUEUE_NAME, false, consumer);

    System.out.println(" [x] Awaiting RPC requests");

    while (true) {
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();

        BasicProperties props = delivery.getProperties();
        BasicProperties replyProps = new BasicProperties.Builder()
                .correlationId(props.getCorrelationId()).build();

        String message = new String(delivery.getBody());
        int n = Integer.parseInt(message);

        System.out.println(" [.] fib(" + message + ")");
        String response = "" + fib(n);

        channel.basicPublish("", props.getReplyTo(), replyProps,
                response.getBytes());

        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
}

/**
 * Constructor.
 *
 * @param exchangeName            - exchangeName
 * @param maxLength               - maxLength
 * @param messageExpiry           - messageExpiry
 * @param networkRecoveryInterval - networkRecoveryInterval
 * @param messageTtl              - messageTtl
 * @param routingHeader           - routingHeader
 */
private RabbitMQConfig(String exchangeName, int maxLength,
                       int messageExpiry, int networkRecoveryInterval,
                       int messageTtl, BasicProperties routingHeader) {
    this.exchangeName = exchangeName;
    this.maxLength = maxLength;
    this.messageExpiry = messageExpiry;
    this.networkRecoveryInterval = networkRecoveryInterval;
    this.messageTtl = messageTtl;
    this.routingHeader = routingHeader;
}

@Test
public void testConnectionFactory() throws Exception {
    Assert.assertNotNull(connectionFactory1);
    Assert.assertNotNull(queue);

    RabbitmqConnection connection = connectionFactory1.getConnection();
    Assert.assertNotNull(connection);
    String queueName = "testing";
    Channel channel = connection.createChannel();
    channel.queueDeclare(queueName, false, false, false, null);
    String message = "Hello World!";

    final CountDownLatch counter = new CountDownLatch(1);
    Consumer consume = new DefaultConsumer(channel) {
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope,
                BasicProperties properties, byte[] body) throws IOException {
            Assert.assertEquals("Hello World!", new String(body));
            counter.countDown();
        }
    };

    channel.basicConsume(queueName, true, consume);
    channel.basicPublish("", queueName, null, message.getBytes());
    counter.await(10, TimeUnit.SECONDS);
    Assert.assertEquals(0, counter.getCount());
    channel.close();

}