RabbitMQTaskQueue implements TaskQueue { protected Task marshallTask(byte[] msgBody, long deliveryTag, String routingKey, String exchange) { Properties props = new Properties(); Task task = null; String msg = new String(msgBody); try { props.load(new StringReader(msg)); if (props.containsKey(Task.KEY_TYPE)) { task = new Task(); for (final String key : props.stringPropertyNames()) { if (key.equals(Task.KEY_TYPE)) { task.setType(Task.Type.valueOf(props.getProperty(key))); } else { task.addProperty(key, props.getProperty(key)); } } task.addProperty(MsgProp.DELIVERY_TAG.name(), String.valueOf(deliveryTag)); task.addProperty(MsgProp.ROUTING_KEY.name(), routingKey); task.addProperty(MsgProp.EXCHANGE.name(), exchange); } else { log.error("RabbitMQ message from queue: " + queueName + " at " + queueUrl + ", does not contain a 'task type'"); } } catch (IOException ioe) { log.error("Error creating Task", ioe); } return task; } RabbitMQTaskQueue(String host, Integer port, String vhost, String exchange, String username, String password,
                             String queueName);  RabbitMQTaskQueue(Connection conn, String exchange, String queueName); @Override String getName(); @Override void put(final Task task); @Override void put(Task... tasks); @Override void put(Set<Task> tasks); @Override Set<Task> take(int maxTasks); @Override Task take(); @Override void extendVisibilityTimeout(Task task); @Override void deleteTask(Task task); @Override void deleteTasks(Set<Task> tasks); @Override void requeue(Task task); @Override Integer size(); @Override Integer sizeIncludingInvisibleAndDelayed();  }
@Test public void testMarshallTask() throws IOException { setupRabbitMQClient(); replayAll(); createSubject(); String message = Task.KEY_TYPE + "=" + Task.Type.DUP.name() + "\n" + "key1=value1\nkey2=value2\nkey3=value3"; byte[] msgBody = message.getBytes(); Task task = queue.marshallTask(msgBody, 0, queueName, exchange); assertThat(task.getProperty(RabbitMQTaskQueue.MsgProp.DELIVERY_TAG.name()), is(equalTo("0"))); assertThat(task.getProperty(RabbitMQTaskQueue.MsgProp.EXCHANGE.name()), is(equalTo("test-exchange"))); assertThat(task.getProperty(RabbitMQTaskQueue.MsgProp.ROUTING_KEY.name()), is(equalTo("test-queue"))); assertThat(task.getType(), is(equalTo(Task.Type.DUP))); assertThat(task.getProperty("key1"), is(equalTo("value1"))); assertThat(task.getProperty("key2"), is(equalTo("value2"))); assertThat(task.getProperty("key3"), is(equalTo("value3"))); }