Java 类com.rabbitmq.client.MessageProperties 实例源码

/**
 * @param args
 * @throws IOException
 * @throws TimeoutException
 * @date 2017年7月11日 下午5:53:02
 * @writer junehappylove
 */
public static void main(String[] args) throws IOException, TimeoutException {
    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(TASK_QUEUE_NAME, true, false, false, null);
    // 分发信息
    for (int i = 0; i < 20; i++) {
        String message = "Hello RabbitMQ" + i;
        channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
        System.out.println("NewTask send '" + message + "'");
    }
    channel.close();
    connection.close();
}

public static void main(String[] argv) throws Exception {
  ConnectionFactory factory = new ConnectionFactory();
  factory.setHost("localhost");
  Connection connection = factory.newConnection();
  Channel channel = connection.createChannel();

  channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

  String message = getMessage(argv);

  channel.basicPublish("", TASK_QUEUE_NAME,
      MessageProperties.PERSISTENT_TEXT_PLAIN,
      message.getBytes("UTF-8"));
  System.out.println(" [x] Sent '" + message + "'");

  channel.close();
  connection.close();
}

/****
 * 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);
}

private void publishEventToExchange(final Channel channel, final String exchangeName, final String routingKey, final String eventDescription) throws IOException {
    final byte[] messageBodyBytes = eventDescription.getBytes();
    try {
        channel.basicPublish(exchangeName, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);
    } catch (final IOException e) {
        s_logger.warn("Failed to publish event " + routingKey + " on exchange " + exchangeName + "  of message broker due to " + e.getMessage(), e);
        throw e;
    }
}

public static void main(String[] argv) throws java.io.IOException, Exception {

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
//      �ַ���Ϣ
        for(int i = 0 ; i < 5; i++){
            String message = "Hello World! " + i;
            channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
        }
        channel.close();
        connection.close();
    }

@Test
public void test_topic() {
  Topic topic = Topic.builder()
      .durable(false)
      .exchange("exchange")
      .autoDelete(false)
      .exclusive(false)
      .name("name")
      .properties(MessageProperties.PERSISTENT_TEXT_PLAIN)
      .build();

  assertTrue(!topic.durable());
  assertTrue(!topic.autoDelete());
  assertTrue(!topic.exclusive());
  assertEquals(topic.exchange(), "exchange");
  assertEquals(topic.name(), "name");
  assertEquals(topic.properties(), MessageProperties.PERSISTENT_TEXT_PLAIN);


}

private static AMQP.BasicProperties convert(String name) throws RuleException {
    switch (name) {
        case "BASIC":
            return MessageProperties.BASIC;
        case "TEXT_PLAIN":
            return MessageProperties.TEXT_PLAIN;
        case "MINIMAL_BASIC":
            return MessageProperties.MINIMAL_BASIC;
        case "MINIMAL_PERSISTENT_BASIC":
            return MessageProperties.MINIMAL_PERSISTENT_BASIC;
        case "PERSISTENT_BASIC":
            return MessageProperties.PERSISTENT_BASIC;
        case "PERSISTENT_TEXT_PLAIN":
            return MessageProperties.PERSISTENT_TEXT_PLAIN;
        default:
            throw new RuleException("Message Properties: '" + name + "' is undefined!");
    }
}

public static void main(String[] args) throws IOException, TimeoutException {

        //建立连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置连接地址
        factory.setHost("seaof-153-125-234-173.jp-tokyo-10.arukascloud.io");
        factory.setPort(31084);
        //获取连接
        Connection connection = factory.newConnection();
        //获取渠道
        Channel channel = connection.createChannel();
        //声明队列，如果不存在就新建
        //参数1队列名称;参数2是否持久化;参数3排他性队列，连接断开自动删除;参数4是否自动删除;参数5.参数
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //发送的消息
        for (int i = 0; i < 10; i++) {

            String message = "Hello";
            int num = ThreadLocalRandom.current().nextInt(10);
            String append = "";
            for (int j = 0; j < num; j++) {
                append = append + ".";
            }
            message = message + (append);

            //参数1 交换机；参数2 路由键；参数3 基础属性,（持久化方式）；参数4 消息体
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
            System.out.println(Thread.currentThread().getName() + "[send]" + message);
        }
        channel.close();
        connection.close();


    }

/**
 * Send messages over queue and DB.
 * @param request request from the client
 */
public String send(Request request, String clientIp) {
    Measurement measurement =
            new Measurement(request.getClientId(), clientIp, request.getCommand(),
                    request.getResponseAddress(),request.getRepetitions(), request.getRepetitionInterval(),
                    request.getProcessors(), request.getAdapter());
    String measurementString = JsonConverter.objectToJsonString(measurement);

    //put data in DB
    String id = mm.pushJson(measurementString);
    MDC.put("jobId", id);
    logger.info("Json pushed in DB: " + measurementString);
    MDC.remove("jobId");

    try {
        channel.basicPublish("", queueName,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                JsonConverter.objectToJsonString(new Job(id)).getBytes());
    } catch (IOException e) {
        logger.error(e.getMessage(), e);
    }
    return id;
}

public void sendMessage(File messageFile) throws IOException, ShutdownSignalException, InterruptedException{
    InputStream is = new FileInputStream(messageFile);

    // Get the size of the file
    long length = messageFile.length();
    if (length > Integer.MAX_VALUE) {
        throw new IOException("Input File ("+messageFile.getName()+") is to large! ");
    }
    byte[] messageBodyBytes = new byte[(int)length];
    int offset = 0;
    int numRead = 0;
    while (offset < messageBodyBytes.length
           && (numRead=is.read(messageBodyBytes, offset, messageBodyBytes.length-offset)) >= 0) {
        offset += numRead;
    }
    if (offset < messageBodyBytes.length) {
        throw new IOException("Could not completely read file "+messageFile.getName());
    }
    is.close();
    this.channel.basicPublish(this.ExchangeName, this.RoutingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes) ;

}

public static void sendMessage(String message)
        throws java.io.IOException {

    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost(ConfigurationLoader.getInstance().getRabbitmqNodename());
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

    channel.basicPublish( "", TASK_QUEUE_NAME,
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            message.getBytes());
    System.out.println(" [x] Sent '" + message + "'");

    channel.close();
    connection.close();
}

public static void sendMessage(String message)
        throws java.io.IOException {

    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost(ConfigurationLoader.getInstance().getRabbitmqNodename());
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

    channel.basicPublish( "", TASK_QUEUE_NAME,
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            message.getBytes());
    System.out.println(" [x] Sent '" + message + "'");

    channel.close();
    connection.close();
}

public static void main(String[] args) throws Exception {


        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        System.out.print("Down Stream server ip (localhost):");
        String server_ip = br.readLine();
        if (server_ip.equalsIgnoreCase("")) {
            server_ip = "localhost";
        }

        ConnectionFactory factory_down = new ConnectionFactory();
        factory_down.setHost(server_ip);
        factory_down.setUsername("test");
        factory_down.setPassword("test");
        factory_down.setVirtualHost("poc");
        Connection connection_down = factory_down.newConnection();
        System.out.println("Connected to Down Stream node: " + server_ip);
        final Channel channel_down = connection_down.createChannel();
        final String exchange = "service";
        channel_down.exchangeDeclare(exchange, "topic", true);
        channel_down.basicPublish(exchange, "r1.gis", MessageProperties.PERSISTENT_BASIC, "better".getBytes());

    }

private Queue<byte[]> sendInternal(byte[] message) throws IOException {
    try {
        channel.basicPublish(DEFAULT_EXCHANGE, this.queueName, MessageProperties.PERSISTENT_BASIC, message);
    } catch (Exception e) {
        // try to reconnect and re-try once...
        connect();
        channel.basicPublish(DEFAULT_EXCHANGE, this.queueName, MessageProperties.PERSISTENT_BASIC, message);
        // if that fails, it simply throws an exception
    }
    return this;
}

/**
 * 发送消息
 *
 * @param message 消息内容 不能为空
 */
public void sendWorkQueueMessage(String message) throws IOException, TimeoutException {
    RabbitMQChannel channel = new RabbitMQChannel().channel();
    System.out.println(channel);
    channel.getChannel().queueDeclare(WORK_QUEUE_NAME, true, false, false, null);
    channel.getChannel().basicPublish("", WORK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    System.out.println("发送workQueue消息:" + message);
    channel.close();
}

private void sendQueue(QueueName queueName, RabbitMessage rm) throws Exception {
    if (rm == null || queueName == null) {
        return;
    }
    initQueueChannel();
    String _queueName = queueName.getNameByEnvironment(environment);
    Transaction trans = Cat.newTransaction("RabbitMQ Message", "PUBLISH-QUEUE-" + _queueName);
    Cat.logEvent("mq send queue", _queueName, Event.SUCCESS,rm.toJsonStr());
    try {
        queueChannel.queueDeclare(_queueName, true, false, false, null);
        queueChannel.basicPublish("", _queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, rm.toJsonStr().getBytes("UTF-8"));
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("SEND SUCCESS:[queue:{},message:{}]", _queueName, rm.toJsonStr());
        }
        Cat.logMetricForCount("PUBLISH-QUEUE-" + _queueName); // 统计请求次数, 可以查看对应队列中放入了多少信息
        trans.setStatus(Transaction.SUCCESS);
    } catch (Exception e) {
        if (LOGGER.isErrorEnabled()) {
            LOGGER.error("SEND ERROR:[queue:{},message:{},exception:{}]", _queueName, rm.toJsonStr(), e);
        }
        String err = queueName + "  rabbitmq发送消息异常";
        Cat.logError(err, e);
        trans.setStatus(e);
        throw new AsuraRabbitMqException(err, e);
    } finally {
        trans.complete();
    }
}

public void send(JobRequestMessage requestMessage, String queueName) {
    try {
        if(!queueNamesCache.contains(queueName)) {
            channel.queueDeclare(queueName, true, false, false, null);
            queueNamesCache.add(queueName);
        }
        logger.info(String.format("Produce job to queue %s", queueName));

        String sendMessage = requestMessage.toJson().toString();
        channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, sendMessage.getBytes());
    } catch (IOException e) {
        logger.error("Error publish job message", e);
    }
}

@Override
public void publish(String jsonMessage, String routingKey) {
    Validate.notNull(routingKey, "routing key is required");
    Charset charset = amqpBrokerConfig.getCharset();

    try {
        channel.basicPublish(exchangeName, routingKey, MessageProperties.PERSISTENT_BASIC, jsonMessage.getBytes(charset));
    } catch (Exception e) {
        LOG.error(ERROR_MESSAGE_TEMPLATE, exchangeName, routingKey);
        LOG.trace("Message: {}", jsonMessage);
        throw new ChannelException(String.format(ERROR_MESSAGE_TEMPLATE, exchangeName, routingKey), e);
    }
}

@Test
public void testPublish() throws ChannelException, IOException {
    String message = "message";
    AmqpProducerAdapter producerAdapter = new AmqpProducerAdapter(TOPIC_NAME, ExchangeType.FANOUT, mockAmqpBrokerConfig, mockAmqpConnectionManager);
    producerAdapter.publish(message);
    verify(mockChannel).basicPublish(TOPIC_NAME, StringUtils.EMPTY, MessageProperties.PERSISTENT_BASIC, message.getBytes());
}

@Test
public void testPublishWithRoutingKey() throws Exception{
    String message = "message";
    String routingKey = "routingKey";
    AmqpProducerAdapter producerAdapter = new AmqpProducerAdapter(TOPIC_NAME, ExchangeType.TOPIC, mockAmqpBrokerConfig, mockAmqpConnectionManager);

    producerAdapter.publish(message, routingKey);
    verify(mockChannel).basicPublish(TOPIC_NAME, routingKey, MessageProperties.PERSISTENT_BASIC, message.getBytes());
}

@Test
public void sendPersistent() throws Exception {
    String QUEUE_NAME = "TEST_PERSISTENT";
    //
    String message = "Hello World";
    DeclareOk declare = channel.queueDeclare(QUEUE_NAME, true, false,
            false, null);// durable=true
    // System.out.println("declare: " + declare);
    channel.basicPublish("", QUEUE_NAME,
            MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    System.out.println(" [x] Sent '" + message + "'");
}

/**
 * send job over the queue
 *
 * @param job contains details about the job
 */
public void send(Job job) {

    try {
        channel.basicPublish(EXCHANGE_NAME, "",
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                JsonConverter.objectToJsonString(job).getBytes());
    } catch (IOException e) {
        logger.error(e.getMessage(), e);
    }

}

/**
 * send job over the queue
 *
 * @param job contains details about the job
 */
public void send(Job job) {
    try {
        channel.basicPublish("", queueName,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                JsonConverter.objectToJsonString(job).getBytes());
    } catch (IOException e) {
        logger.error(e.getMessage(), e);
    }
}

/**
 * send job over the queue
 *
 * @param job contains details about the job
 */
public void send(Job job) {
    try {
        channel.basicPublish("", queueName,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                JsonConverter.objectToJsonString(job).getBytes());
    } catch (IOException e) {
        logger.error(e.getMessage(), e);
    }
}

/**
 * send job over the queue
 *
 * @param job contains details about the job
 */
public void send(Job job) {
    try {
        channel.basicPublish("", queueName,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                JsonConverter.objectToJsonString(job).getBytes());
    } catch (IOException e) {
        logger.error(e.getMessage(), e);
    }
}

public void testExFanoutSending() throws Exception {

    RabbitMQExchanger exchanger = new RabbitMQExchanger("127.0.0.1", 5672, "order1", BuiltinExchangeType.FANOUT);
    for (int i=0; i<10; i++) {
        exchanger.send("hello: "+i, MessageProperties.MINIMAL_PERSISTENT_BASIC);
        Thread.sleep(100);
    }
    exchanger.close();
}

public void testExTopicSending() throws Exception {

    RabbitMQExchanger exchanger = new RabbitMQExchanger("127.0.0.1", 5672, "order", BuiltinExchangeType.TOPIC);
    for (int i=0; i<10; i++) {
        exchanger.send(("hello: "+i).getBytes(), ((i%2==0)?"topicA.hahah":"topicB.hahah"), MessageProperties.MINIMAL_PERSISTENT_BASIC);
        Thread.sleep(100);
    }
    exchanger.close();
}

@Override
public boolean publishStream(StreamID streamID) {
    String msg = RegistryProtocol.PUBLISH.toString() + " " + streamID.getRoute();
    try {
        channel.basicPublish("", registryQueue,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                msg.getBytes());
        return true;
    } catch (IOException e) {
        e.printStackTrace();
        return false;
    }
}

@Override
public boolean unpublishStream(StreamID streamID) {
    String msg = RegistryProtocol.UNPUBLISH.toString() + " " + streamID.getRoute();
    try {
        channel.basicPublish("", registryQueue,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                msg.getBytes());
        return true;
    } catch (IOException e) {
        e.printStackTrace();
        return false;
    }
}

public void send(byte []message, String routingKey) throws Exception {
    try {
        channel.basicPublish(properties.getExchangeName(), routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message);
    } catch (IOException e) {
        String msg = "Failed to publish message to exchange: " + properties.getExchangeName();
        log.error(msg, e);
        throw new Exception(msg, e);
    }
}

private void publishEventToExchange(Channel channel, String exchangeName, String routingKey, String eventDescription) throws Exception {
    try {
        byte[] messageBodyBytes = eventDescription.getBytes();
        channel.basicPublish(exchangeName, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);
    } catch (Exception e) {
        s_logger.error("Failed to publish event " + routingKey + " on exchange " + exchangeName + "  of message broker due to " + e.getMessage());
        throw e;
    }
}

@Override
public void publish(T message) {
    try {
        channel.basicPublish("", name, MessageProperties.PERSISTENT_TEXT_PLAIN, MAPPER.writeValueAsBytes(message));

        publish.mark();
        if (log.isTraceEnabled()) {
            log.trace("Published to '{}' with data '{}'.", name, MAPPER.writeValueAsString(message));
        }
    } catch (IOException e) {
        throw new MessageQueueException("Unable to publish to queue.", e);
    }
}

@Test
public void shouldPublishPersistentMessageToQueue() throws IOException {
    RabbitMQMessageQueue<TestMsg> queue = new RabbitMQMessageQueue<>(channel, QUEUE_NAME, TestMsg.class, metrics);
    TestMsg message = new TestMsg("blah", 5);
    byte[] messageBytes = Jackson.newObjectMapper().writeValueAsBytes(message);

    queue.publish(message);

    verify(channel).basicPublish(Matchers.eq(""), Matchers.eq(QUEUE_NAME), Matchers.eq(MessageProperties.PERSISTENT_TEXT_PLAIN), Matchers.eq(messageBytes));
}

private AMQP.BasicProperties createProps(InternalMessage message) {
    if(message.getTimeout() < 0) {
        return message.isDurable() ? MessageProperties.PERSISTENT_BASIC : MessageProperties.BASIC;
    } else {
        if(message.isDurable()) {
            return new AMQP.BasicProperties.Builder().contentType("application/octet-stream").deliveryMode(2)
                    .priority(0).expiration(String.valueOf(message.getTimeout())).build();
        } else {
            return new AMQP.BasicProperties.Builder().contentType("application/octet-stream").deliveryMode(1)
                    .priority(0).expiration(String.valueOf(message.getTimeout())).build();
        }
    }
}

private AMQP.BasicProperties createProps(InternalMessage message) {
    if(message.getTimeout() < 0) {
        return message.isDurable() ? MessageProperties.PERSISTENT_BASIC : MessageProperties.BASIC;
    } else {
        if(message.isDurable()) {
            return new AMQP.BasicProperties.Builder().contentType("application/octet-stream").deliveryMode(2)
                    .priority(0).expiration(String.valueOf(message.getTimeout())).build();
        } else {
            return new AMQP.BasicProperties.Builder().contentType("application/octet-stream").deliveryMode(1)
                    .priority(0).expiration(String.valueOf(message.getTimeout())).build();
        }
    }
}

/**
 * Handles incoming command request from the hobbit command queue.
 *
 * <p>
 * Commands handled by this method:
 * <ul>
 * <li>{@link Commands#DOCKER_CONTAINER_START}</li>
 * <li>{@link Commands#DOCKER_CONTAINER_STOP}</li>
 * </ul>
 *
 * @param command
 *            command to be executed
 * @param data
 *            byte-encoded supplementary json for the command
 *
 *            0 - start container 1 - stop container Data format for each
 *            command: Start container:
 */
public void receiveCommand(byte command, byte[] data, String sessionId, String replyTo) {
    if (LOGGER.isDebugEnabled()) {
        LOGGER.info("received command: session={}, command={}, data={}", sessionId, Commands.toString(command),
                data != null ? RabbitMQUtils.readString(data) : "null");
    } else {
        LOGGER.info("received command: session={}, command={}", sessionId, Commands.toString(command));
    }
    // This command will receive data from Rabbit
    // determine the command
    switch (command) {
    case Commands.DOCKER_CONTAINER_START: {
        // Convert data byte array to config data structure
        StartCommandData startParams = deserializeStartCommandData(data);
        // trigger creation
        String containerName = createContainer(startParams);
        if (replyTo != null) {
            try {
                cmdChannel.basicPublish("", replyTo, MessageProperties.PERSISTENT_BASIC,
                        RabbitMQUtils.writeString(containerName));
            } catch (IOException e) {
                StringBuilder errMsgBuilder = new StringBuilder();
                errMsgBuilder.append("Error, couldn't sent response after creation of container (");
                errMsgBuilder.append(startParams.toString());
                errMsgBuilder.append(") to replyTo=");
                errMsgBuilder.append(replyTo);
                errMsgBuilder.append(".");
                LOGGER.error(errMsgBuilder.toString(), e);
            }
        }
        break;
    }
    case Commands.DOCKER_CONTAINER_STOP: {
        // get containerId from params
        StopCommandData stopParams = deserializeStopCommandData(data);
        // trigger stop
        stopContainer(stopParams.containerName);
        break;
    }
    case Commands.BENCHMARK_READY_SIGNAL: {
        expManager.systemOrBenchmarkReady(false);
        break;
    }
    case Commands.SYSTEM_READY_SIGNAL: {
        expManager.systemOrBenchmarkReady(true);
        break;
    }
    case Commands.TASK_GENERATION_FINISHED: {
        expManager.taskGenFinished();
        break;
    }
    case Commands.BENCHMARK_FINISHED_SIGNAL: {
        if ((data == null) || (data.length == 0)) {
            LOGGER.error("Got no result model from the benchmark controller.");
        } else {
            Model model = RabbitMQUtils.readModel(data);
            expManager.setResultModel(model);
        }
        break;
    }
    }
}

@Override
public void analyzeExperiment(String uri) throws IOException {
    controller2Analysis.basicPublish("", Constants.CONTROLLER_2_ANALYSIS_QUEUE_NAME,
            MessageProperties.PERSISTENT_BASIC, RabbitMQUtils.writeString(uri));
}

private void send(MQItem item) throws IOException {
    channel.basicPublish("", this.queueName, 
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            gson.toJson(item).getBytes());
}

public static ExchangeSignal from(String body) {
    return from(body.getBytes(), MessageProperties.TEXT_PLAIN);
}

public static ExchangeSignal route(String body, String routingKey) {
    return route(body.getBytes(), routingKey, null, MessageProperties.TEXT_PLAIN);
}