@Test void workQueueShouldConsumeSubmittedTask() { testee.submit(TASK_WITH_ID); await().atMost(FIVE_HUNDRED_MILLISECONDS).until(() -> !worker.results.isEmpty()); assertThat(worker.tasks).containsExactly(TASK_WITH_ID); assertThat(worker.results).containsExactly(Task.Result.COMPLETED); }
@Override public void submit(TaskWithId taskWithId) { try { byte[] payload = taskSerializer.serialize(taskWithId.getTask()).getBytes(StandardCharsets.UTF_8); AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder() .deliveryMode(PERSISTENT_TEXT_PLAIN.getDeliveryMode()) .priority(PERSISTENT_TEXT_PLAIN.getPriority()) .contentType(PERSISTENT_TEXT_PLAIN.getContentType()) .headers(ImmutableMap.of(TASK_ID, taskWithId.getId().asString())) .build(); OutboundMessage outboundMessage = new OutboundMessage(EXCHANGE_NAME, ROUTING_KEY, basicProperties, payload); sender.send(Mono.just(outboundMessage)).block(); } catch (JsonProcessingException e) { throw new RuntimeException(e); } }
RabbitMQWorkQueue implements WorkQueue { @Override public void submit(TaskWithId taskWithId) { try { byte[] payload = taskSerializer.serialize(taskWithId.getTask()).getBytes(StandardCharsets.UTF_8); AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder() .deliveryMode(PERSISTENT_TEXT_PLAIN.getDeliveryMode()) .priority(PERSISTENT_TEXT_PLAIN.getPriority()) .contentType(PERSISTENT_TEXT_PLAIN.getContentType()) .headers(ImmutableMap.of(TASK_ID, taskWithId.getId().asString())) .build(); OutboundMessage outboundMessage = new OutboundMessage(EXCHANGE_NAME, ROUTING_KEY, basicProperties, payload); sender.send(Mono.just(outboundMessage)).block(); } catch (JsonProcessingException e) { throw new RuntimeException(e); } } }
RabbitMQWorkQueue implements WorkQueue { @Override public void submit(TaskWithId taskWithId) { try { byte[] payload = taskSerializer.serialize(taskWithId.getTask()).getBytes(StandardCharsets.UTF_8); AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder() .deliveryMode(PERSISTENT_TEXT_PLAIN.getDeliveryMode()) .priority(PERSISTENT_TEXT_PLAIN.getPriority()) .contentType(PERSISTENT_TEXT_PLAIN.getContentType()) .headers(ImmutableMap.of(TASK_ID, taskWithId.getId().asString())) .build(); OutboundMessage outboundMessage = new OutboundMessage(EXCHANGE_NAME, ROUTING_KEY, basicProperties, payload); sender.send(Mono.just(outboundMessage)).block(); } catch (JsonProcessingException e) { throw new RuntimeException(e); } } RabbitMQWorkQueue(TaskManagerWorker worker, Sender sender, ReceiverProvider receiverProvider, JsonTaskSerializer taskSerializer); }
RabbitMQWorkQueue implements WorkQueue { @Override public void submit(TaskWithId taskWithId) { try { byte[] payload = taskSerializer.serialize(taskWithId.getTask()).getBytes(StandardCharsets.UTF_8); AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder() .deliveryMode(PERSISTENT_TEXT_PLAIN.getDeliveryMode()) .priority(PERSISTENT_TEXT_PLAIN.getPriority()) .contentType(PERSISTENT_TEXT_PLAIN.getContentType()) .headers(ImmutableMap.of(TASK_ID, taskWithId.getId().asString())) .build(); OutboundMessage outboundMessage = new OutboundMessage(EXCHANGE_NAME, ROUTING_KEY, basicProperties, payload); sender.send(Mono.just(outboundMessage)).block(); } catch (JsonProcessingException e) { throw new RuntimeException(e); } } RabbitMQWorkQueue(TaskManagerWorker worker, Sender sender, ReceiverProvider receiverProvider, JsonTaskSerializer taskSerializer); @Override void start(); @Override void submit(TaskWithId taskWithId); @Override void cancel(TaskId taskId); @Override void close(); }
RabbitMQWorkQueue implements WorkQueue { @Override public void submit(TaskWithId taskWithId) { try { byte[] payload = taskSerializer.serialize(taskWithId.getTask()).getBytes(StandardCharsets.UTF_8); AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder() .deliveryMode(PERSISTENT_TEXT_PLAIN.getDeliveryMode()) .priority(PERSISTENT_TEXT_PLAIN.getPriority()) .contentType(PERSISTENT_TEXT_PLAIN.getContentType()) .headers(ImmutableMap.of(TASK_ID, taskWithId.getId().asString())) .build(); OutboundMessage outboundMessage = new OutboundMessage(EXCHANGE_NAME, ROUTING_KEY, basicProperties, payload); sender.send(Mono.just(outboundMessage)).block(); } catch (JsonProcessingException e) { throw new RuntimeException(e); } } RabbitMQWorkQueue(TaskManagerWorker worker, Sender sender, ReceiverProvider receiverProvider, JsonTaskSerializer taskSerializer); @Override void start(); @Override void submit(TaskWithId taskWithId); @Override void cancel(TaskId taskId); @Override void close(); static final String TASK_ID; static final int NUM_RETRIES; static final Duration FIRST_BACKOFF; }