RabbitMQTaskQueue implements TaskQueue { protected String unmarshallTask(Task task) { Properties props = new Properties(); props.setProperty(Task.KEY_TYPE, task.getType().name()); for (String key : task.getProperties().keySet()) { String value = task.getProperty(key); if (null != value) { props.setProperty(key, value); } } StringWriter sw = new StringWriter(); String msgBody = null; try { props.store(sw, null); msgBody = sw.toString(); } catch (IOException ioe) { log.error("Error unmarshalling Task, queue: " + queueName + ", msgBody: " + msgBody, ioe); } return msgBody; } 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 testUnmarshallTask() throws IOException { setupRabbitMQClient(); replayAll(); createSubject(); Task task = new Task(); task.setType(Task.Type.DUP); task.addProperty("key1", "value1"); task.addProperty("key2", "value2"); task.addProperty("key3", "value3"); String msgBody = queue.unmarshallTask(task); Assert.assertTrue(msgBody.contains("key1=value1")); Assert.assertTrue(msgBody.contains("key2=value2")); Assert.assertTrue(msgBody.contains("key3=value3")); }