@Test void headersAreAddedToTheTaskRequest() { Multimap<String, byte[]> headers = HashMultimap.create(); headers.put("a", "b".getBytes(UTF_8)); toKafkaSenderService.sendMessage(new SendMessageRequest() .setPayload("abc") .setHeaders(headers)); ArgumentCaptor<AddTaskRequest> taskCaptor = ArgumentCaptor.forClass(AddTaskRequest.class); verify(taskService).addTask(taskCaptor.capture()); AddTaskRequest actualRequest = taskCaptor.getValue(); assertHeadersEqual(headers, actualRequest); }
@Override @Transactional(rollbackFor = Exception.class) public void sendMessage(SendMessageRequest request) { ToKafkaMessages messages = new ToKafkaMessages().setTopic(request.getTopic()).add(convert(request)); taskService.addTask(new ITasksService.AddTaskRequest().setType(ToKafkaTaskType.VALUE).setSubType(messages.getTopic()).setData(messages) .setRunAfterTime(request.getSendAfterTime())); }
ToKafkaSenderService implements IToKafkaSenderService { @Override @Transactional(rollbackFor = Exception.class) public void sendMessage(SendMessageRequest request) { ToKafkaMessages messages = new ToKafkaMessages().setTopic(request.getTopic()).add(convert(request)); taskService.addTask(new ITasksService.AddTaskRequest().setType(ToKafkaTaskType.VALUE).setSubType(messages.getTopic()).setData(messages) .setRunAfterTime(request.getSendAfterTime())); } }
ToKafkaSenderService implements IToKafkaSenderService { @Override @Transactional(rollbackFor = Exception.class) public void sendMessage(SendMessageRequest request) { ToKafkaMessages messages = new ToKafkaMessages().setTopic(request.getTopic()).add(convert(request)); taskService.addTask(new ITasksService.AddTaskRequest().setType(ToKafkaTaskType.VALUE).setSubType(messages.getTopic()).setData(messages) .setRunAfterTime(request.getSendAfterTime())); }  }
ToKafkaSenderService implements IToKafkaSenderService { @Override @Transactional(rollbackFor = Exception.class) public void sendMessage(SendMessageRequest request) { ToKafkaMessages messages = new ToKafkaMessages().setTopic(request.getTopic()).add(convert(request)); taskService.addTask(new ITasksService.AddTaskRequest().setType(ToKafkaTaskType.VALUE).setSubType(messages.getTopic()).setData(messages) .setRunAfterTime(request.getSendAfterTime())); }  @Override @Transactional(rollbackFor = Exception.class) void sendMessage(SendMessageRequest request); @Override @Transactional(rollbackFor = Exception.class) void sendMessages(SendMessagesRequest request); }
ToKafkaSenderService implements IToKafkaSenderService { @Override @Transactional(rollbackFor = Exception.class) public void sendMessage(SendMessageRequest request) { ToKafkaMessages messages = new ToKafkaMessages().setTopic(request.getTopic()).add(convert(request)); taskService.addTask(new ITasksService.AddTaskRequest().setType(ToKafkaTaskType.VALUE).setSubType(messages.getTopic()).setData(messages) .setRunAfterTime(request.getSendAfterTime())); }  @Override @Transactional(rollbackFor = Exception.class) void sendMessage(SendMessageRequest request); @Override @Transactional(rollbackFor = Exception.class) void sendMessages(SendMessagesRequest request);  }