@Test void testKafka() { when(mockEvent.getObject()).thenReturn(of(rdf.createIRI("trellis:data/resource"))); when(mockEvent.getAgents()).thenReturn(singleton(Trellis.AdministratorAgent)); when(mockEvent.getIdentifier()).thenReturn(rdf.createIRI("urn:test")); when(mockEvent.getCreated()).thenReturn(time); when(mockEvent.getTypes()).thenReturn(singleton(AS.Update)); when(mockEvent.getObjectTypes()).thenReturn(singleton(LDP.RDFSource)); when(mockEvent.getInbox()).thenReturn(empty()); final EventService svc = new KafkaEventService(serializer, producer); svc.emit(mockEvent); final List<ProducerRecord<String, String>> records = producer.history(); assertEquals(1L, records.size(), "Incorrect total records size!"); assertEquals(1L, records.stream().filter(r -> r.topic().equals(queueName)).count(), "Incorrect filtered size!"); }
@Override public void emit(final Event event) { requireNonNull(event, "Cannot emit a null event!"); LOGGER.debug("Sending message to Kafka topic: {}", topic); producer.send(new ProducerRecord<>(topic, event.getObject().map(IRI::getIRIString).orElse(null), serializer.serialize(event))); }
KafkaEventService implements EventService { @Override public void emit(final Event event) { requireNonNull(event, "Cannot emit a null event!"); LOGGER.debug("Sending message to Kafka topic: {}", topic); producer.send(new ProducerRecord<>(topic, event.getObject().map(IRI::getIRIString).orElse(null), serializer.serialize(event))); } }
KafkaEventService implements EventService { @Override public void emit(final Event event) { requireNonNull(event, "Cannot emit a null event!"); LOGGER.debug("Sending message to Kafka topic: {}", topic); producer.send(new ProducerRecord<>(topic, event.getObject().map(IRI::getIRIString).orElse(null), serializer.serialize(event))); } KafkaEventService(); @Inject  KafkaEventService(final EventSerializationService serializer); private  KafkaEventService(final EventSerializationService serializer, final Config config);  KafkaEventService(final EventSerializationService serializer, final Producer<String, String> producer);  KafkaEventService(final EventSerializationService serializer, final Producer<String, String> producer,
            final String topic); }
KafkaEventService implements EventService { @Override public void emit(final Event event) { requireNonNull(event, "Cannot emit a null event!"); LOGGER.debug("Sending message to Kafka topic: {}", topic); producer.send(new ProducerRecord<>(topic, event.getObject().map(IRI::getIRIString).orElse(null), serializer.serialize(event))); } KafkaEventService(); @Inject  KafkaEventService(final EventSerializationService serializer); private  KafkaEventService(final EventSerializationService serializer, final Config config);  KafkaEventService(final EventSerializationService serializer, final Producer<String, String> producer);  KafkaEventService(final EventSerializationService serializer, final Producer<String, String> producer,
            final String topic); @Override void emit(final Event event); }
KafkaEventService implements EventService { @Override public void emit(final Event event) { requireNonNull(event, "Cannot emit a null event!"); LOGGER.debug("Sending message to Kafka topic: {}", topic); producer.send(new ProducerRecord<>(topic, event.getObject().map(IRI::getIRIString).orElse(null), serializer.serialize(event))); } KafkaEventService(); @Inject  KafkaEventService(final EventSerializationService serializer); private  KafkaEventService(final EventSerializationService serializer, final Config config);  KafkaEventService(final EventSerializationService serializer, final Producer<String, String> producer);  KafkaEventService(final EventSerializationService serializer, final Producer<String, String> producer,
            final String topic); @Override void emit(final Event event); static final String CONFIG_KAFKA_TOPIC; }