RabbitMQTaskQueue implements TaskQueue { @Override public void put(final Task task) { try { String queueName = this.queueName; final String msgBody = unmarshallTask(task); byte[] messageBodyBytes = msgBody.getBytes(); new Retrier(4, 10000, 2).execute(new Retriable() { @Override public Object retry() throws Exception { mqChannel.basicPublish(exchangeName, queueName, null, messageBodyBytes); return null; } }); unAcknowlededMesageCount++; log.info("RabbitMQ message successfully placed {} on queue - queue: {}", task, queueName); } catch (Exception ex) { log.error("failed to place {} on {} at {} due to {}", task, queueName, queueUrl, ex.getMessage()); throw new DuraCloudRuntimeException(ex); } } 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 testPut() throws IOException { setupRabbitMQClient(); replayAll(); createSubject(); Task task = createSampleAuditTask(System.currentTimeMillis()); this.queue.put(task); }
@Test public void testPutMuliple() throws IOException { setupRabbitMQClient(); replayAll(); createSubject(); Set<Task> tasks = new HashSet<>(); long time = System.currentTimeMillis(); for (int i = 0; i < 11; i++) { tasks.add(createSampleAuditTask(time + i)); } this.queue.put(tasks); }