@Test void tasksShouldBeConsumedSequentially() { AtomicLong counter = new AtomicLong(0L); Task task1 = new MemoryReferenceTask(() -> { counter.addAndGet(1); Thread.sleep(1000); return Task.Result.COMPLETED; }); TaskId taskId1 = TaskId.fromString("1111d081-aa30-11e9-bf6c-2d3b9e84aafd"); TaskWithId taskWithId1 = new TaskWithId(taskId1, task1); Task task2 = new MemoryReferenceTask(() -> { counter.addAndGet(2); return Task.Result.COMPLETED; }); TaskId taskId2 = TaskId.fromString("2222d082-aa30-22e9-bf6c-2d3b9e84aafd"); TaskWithId taskWithId2 = new TaskWithId(taskId2, task2); testee.submit(taskWithId1); testee.submit(taskWithId2); assertThatThrownBy(() -> await().atMost(FIVE_HUNDRED_MILLISECONDS).untilAtomic(counter, CoreMatchers.equalTo(3L))).isInstanceOf(ConditionTimeoutException.class); assertThatCode(() -> await().atMost(TWO_SECONDS).untilAtomic(counter, CoreMatchers.equalTo(3L))).doesNotThrowAnyException(); }
@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; }