Skip to content

I
inventory-service
Commit 5061bffd authored by Ben Anderson's avatar Ben Anderson
Browse files
Options

Deployment 57

parent 2062559a
Hide whitespace changes
Inline Side-by-side
Showing with 151 additions and 151 deletions
deployment.yaml
View file @ 5061bffd
...@@ -14,7 +14,7 @@ spec: ...@@ -14,7 +14,7 @@ spec:
spec: spec:
containers: containers:
- name: afp-prods-container - name: afp-prods-container
image: nexus.mynisum.com/afp-prods:56 image: nexus.mynisum.com/afp-prods:57
imagePullPolicy: Always imagePullPolicy: Always
ports: ports:
- containerPort: 8083 - containerPort: 8083
......
src/main/java/com/nisum/ascend/inventory/configuration/KafkaReceiverConfig.java
View file @ 5061bffd
package com.nisum.ascend.inventory.configuration; //package com.nisum.ascend.inventory.configuration;
//
//
import com.nisum.ascend.inventory.dto.Order; //import com.nisum.ascend.inventory.dto.Order;
import com.nisum.ascend.inventory.dto.WareHouseOrder; //import com.nisum.ascend.inventory.dto.WareHouseOrder;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig; //import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig; //import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer; //import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value; //import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean; //import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; //import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.support.serializer.JsonDeserializer; //import org.springframework.kafka.support.serializer.JsonDeserializer;
import reactor.kafka.receiver.KafkaReceiver; //import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions; //import reactor.kafka.receiver.ReceiverOptions;
//
import java.util.Collection; //import java.util.Collection;
import java.util.Collections; //import java.util.Collections;
import java.util.HashMap; //import java.util.HashMap;
import java.util.Map; //import java.util.Map;
//
@Configuration //@Configuration
@Slf4j //@Slf4j
public class KafkaReceiverConfig { //public class KafkaReceiverConfig {
//
@Value("${kafka.consumer.bootstrap-servers}") // @Value("${kafka.consumer.bootstrap-servers}")
private String bootstrapServers; // private String bootstrapServers;
@Value("${kafka.consumer.group-id}") // @Value("${kafka.consumer.group-id}")
private String groupId; // private String groupId;
//
@Bean("kafkaWarehouseOrderReceiver") // @Bean("kafkaWarehouseOrderReceiver")
public KafkaReceiver<String, String> kafkaWarehouseOrderEventReceiver( // public KafkaReceiver<String, String> kafkaWarehouseOrderEventReceiver(
@Value("${kafka.WAREHOUSETOPIC.input}") String posLogTopic) { // @Value("${kafka.WAREHOUSETOPIC.input}") String posLogTopic) {
ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(WarehouseOrderEventReceiverConfig()); // ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(WarehouseOrderEventReceiverConfig());
receiverOptions.maxCommitAttempts(3); // receiverOptions.maxCommitAttempts(3);
return KafkaReceiver.create(receiverOptions.addAssignListener(Collection::iterator) // return KafkaReceiver.create(receiverOptions.addAssignListener(Collection::iterator)
.subscription(Collections.singleton(posLogTopic))); // .subscription(Collections.singleton(posLogTopic)));
} // }
//
private Map<String, Object> WarehouseOrderEventReceiverConfig() { // private Map<String, Object> WarehouseOrderEventReceiverConfig() {
Map<String, Object> config = new HashMap<>(); // Map<String, Object> config = new HashMap<>();
config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); // config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); // config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); // config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); // config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
return config; // return config;
} // }
//
@Bean("kafkaOrderReceiver") // @Bean("kafkaOrderReceiver")
public KafkaReceiver<String, String> kafkaOrderEventReceiver( // public KafkaReceiver<String, String> kafkaOrderEventReceiver(
@Value("${kafka.ORDERTOPIC.input}") String posLogTopic) { // @Value("${kafka.ORDERTOPIC.input}") String posLogTopic) {
ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(OrderEventReceiverConfig()); // ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(OrderEventReceiverConfig());
receiverOptions.maxCommitAttempts(3); // receiverOptions.maxCommitAttempts(3);
return KafkaReceiver.create(receiverOptions.addAssignListener(Collection::iterator) // return KafkaReceiver.create(receiverOptions.addAssignListener(Collection::iterator)
.subscription(Collections.singleton(posLogTopic))); // .subscription(Collections.singleton(posLogTopic)));
} // }
//
private Map<String, Object> OrderEventReceiverConfig() { // private Map<String, Object> OrderEventReceiverConfig() {
Map<String, Object> config = new HashMap<>(); // Map<String, Object> config = new HashMap<>();
config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); // config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); // config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); // config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); // config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
return config; // return config;
} // }
//
} //}
\ No newline at end of file \ No newline at end of file
src/main/java/com/nisum/ascend/inventory/service/KafkaListenerService.java
View file @ 5061bffd
package com.nisum.ascend.inventory.service; //package com.nisum.ascend.inventory.service;
//
import com.fasterxml.jackson.databind.ObjectMapper; //import com.fasterxml.jackson.databind.ObjectMapper;
import com.nisum.ascend.inventory.dto.Item; //import com.nisum.ascend.inventory.dto.Item;
import com.nisum.ascend.inventory.dto.Order; //import com.nisum.ascend.inventory.dto.Order;
import com.nisum.ascend.inventory.dto.WareHouseOrder; //import com.nisum.ascend.inventory.dto.WareHouseOrder;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier; //import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.event.ApplicationStartedEvent; //import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.event.EventListener; //import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component; //import org.springframework.stereotype.Component;
import reactor.kafka.receiver.KafkaReceiver; //import reactor.kafka.receiver.KafkaReceiver;
//
import java.util.List; //import java.util.List;
//
@Component //@Component
@Slf4j //@Slf4j
public class KafkaListenerService { //public class KafkaListenerService {
//
@Autowired // @Autowired
@Qualifier("kafkaWarehouseOrderReceiver") // @Qualifier("kafkaWarehouseOrderReceiver")
private KafkaReceiver<String, String> kafkaWarehouseOrderReceiver; // private KafkaReceiver<String, String> kafkaWarehouseOrderReceiver;
//
@Autowired // @Autowired
@Qualifier("kafkaOrderReceiver") // @Qualifier("kafkaOrderReceiver")
private KafkaReceiver<String, String> kafkaOrderReceiver; // private KafkaReceiver<String, String> kafkaOrderReceiver;
//
@Autowired // @Autowired
private ProductService productService; // private ProductService productService;
//
@EventListener(ApplicationStartedEvent.class) // @EventListener(ApplicationStartedEvent.class)
public void consumeWarehouseOrderStatus() { // public void consumeWarehouseOrderStatus() {
kafkaWarehouseOrderReceiver.receive() // kafkaWarehouseOrderReceiver.receive()
.doOnNext(record -> log.info("record: {}", record)) // .doOnNext(record -> log.info("record: {}", record))
.doOnNext(record -> onWarehouseOrderStatusReceived(record.value())) // .doOnNext(record -> onWarehouseOrderStatusReceived(record.value()))
.doOnError(throwable -> System.out.println(throwable.getMessage())) // .doOnError(throwable -> System.out.println(throwable.getMessage()))
.subscribe(); // .subscribe();
} // }
//
private void onWarehouseOrderStatusReceived(String warehouseOrderString) { // private void onWarehouseOrderStatusReceived(String warehouseOrderString) {
try { // try {
ObjectMapper objectMapper = new ObjectMapper(); // ObjectMapper objectMapper = new ObjectMapper();
WareHouseOrder warehouseOrder = objectMapper.readValue(warehouseOrderString, WareHouseOrder.class); // WareHouseOrder warehouseOrder = objectMapper.readValue(warehouseOrderString, WareHouseOrder.class);
List<Item> itemList = warehouseOrder.getOrderItems(); // List<Item> itemList = warehouseOrder.getOrderItems();
String status = warehouseOrder.getStatus(); // String status = warehouseOrder.getStatus();
log.info("Received this data: {}", warehouseOrder); // log.info("Received this data: {}", warehouseOrder);
log.info("recieved this list: {}", itemList); // log.info("recieved this list: {}", itemList);
for (Item item : itemList) { // for (Item item : itemList) {
productService.updateProductInventoryBySkuWareHouse(item.getItemSku(), status, item.getItemQuantity()).subscribe(); // productService.updateProductInventoryBySkuWareHouse(item.getItemSku(), status, item.getItemQuantity()).subscribe();
} // }
} catch (Exception e) { // } catch (Exception e) {
log.error("error", e); // log.error("error", e);
} // }
} // }
//
//
@EventListener(ApplicationStartedEvent.class) // @EventListener(ApplicationStartedEvent.class)
public void consumeOrderStatus() { // public void consumeOrderStatus() {
kafkaOrderReceiver.receive() // kafkaOrderReceiver.receive()
.doOnNext(record -> log.info("record: {}", record)) // .doOnNext(record -> log.info("record: {}", record))
.doOnNext(record -> onOrderStatusReceived(record.value())) // .doOnNext(record -> onOrderStatusReceived(record.value()))
.doOnError(throwable -> System.out.println(throwable.getMessage())) // .doOnError(throwable -> System.out.println(throwable.getMessage()))
.subscribe(); // .subscribe();
} // }
//
private void onOrderStatusReceived(String orderString) { // private void onOrderStatusReceived(String orderString) {
try { // try {
ObjectMapper objectMapper = new ObjectMapper(); // ObjectMapper objectMapper = new ObjectMapper();
Order order = objectMapper.readValue(orderString, Order.class); // Order order = objectMapper.readValue(orderString, Order.class);
List<Item> itemList = order.getOrderItems(); // List<Item> itemList = order.getOrderItems();
String status = order.getOrderStatus(); // String status = order.getOrderStatus();
log.info("Received this data: {}", order); // log.info("Received this data: {}", order);
log.info("recieved this list: {}", itemList); // log.info("recieved this list: {}", itemList);
for (Item item : itemList) { // for (Item item : itemList) {
productService.updateProductInventoryBySkuOrder(item.getItemSku(), status, item.getItemQuantity()).block(); // productService.updateProductInventoryBySkuOrder(item.getItemSku(), status, item.getItemQuantity()).block();
} // }
//
} catch (Exception e) { // } catch (Exception e) {
log.error("error", e); // log.error("error", e);
} // }
} // }
} //}
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