Commit 6b6a3847 authored by Darrick Yong's avatar Darrick Yong

Merge branch 'kafka' into 'master'

Kafka

See merge request !5
parents dcb3abf1 97f2eeda
......@@ -29,6 +29,10 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
......
package com.ascendfinalproject.warehouse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import java.io.IOException;
@Service
public class Consumer {
private final Logger logger = LoggerFactory.getLogger(Consumer.class);
// this is placeholder
@KafkaListener(topics = "fulfilled", groupId = "WAREHOUSE_MANAGEMENT")
public void consume(String message) throws IOException {
logger.info(String.format("#### -> Consumed message -> %s", message));
}
}
package com.ascendfinalproject.warehouse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class Producer {
private static final Logger logger = LoggerFactory.getLogger(Producer.class);
private static final String FULFILLED = "fulfilled";
private static final String CANCELLED = "cancelled";
@Autowired
// publish messages to the topic
private KafkaTemplate<String, String> kafkaTemplate;
public void orderFulfilled(String message) {
logger.info(String.format("#### -> this order is fulfilled -> %s", message));
this.kafkaTemplate.send(FULFILLED, message);
}
public void orderCancelled(String message) {
logger.info(String.format("#### -> this order is cancelled -> %s", message));
this.kafkaTemplate.send(CANCELLED, message);
}
}
......@@ -11,3 +11,4 @@ public class WarehouseApplication {
}
}
package com.ascendfinalproject.warehouse.controllers;
import com.ascendfinalproject.warehouse.Producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping(value = "/kafka")
public class KafkaController {
private final Producer producer;
@Autowired
KafkaController(Producer producer) {
this.producer = producer;
}
@PostMapping(value = "/fulfilled")
public void sendMessageToKafkaTopic(@RequestParam("message") String message) {
this.producer.orderFulfilled(message);
}
}
spring.data.mongodb.uri=mongodb+srv://warehouse1:ascendWarehouseProject@warehouse-cluster.xopll.mongodb.net/myFirstDatabase?retryWrites=true&w=majority
spring.data.mongodb.database=test
\ No newline at end of file
spring.data.mongodb.database=test
server:
port: 9000
spring:
kafka:
consumer:
bootstrap-servers: localhost:9092
group-id: WAREHOUSE_MANAGEMENT
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
producer:
bootstrap-servers: localhost:9092
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment