import com.scallop.api.Scallop;
import com.scallop.api.ScallopBuilder;
import com.scallop.api.datastream.DataStream;
import com.scallop.api.datastream.DataStreamBuilder;
import com.scallop.api.module.Module;
import com.scallop.api.module.ModuleBuilder;
public class OrderProcessingPipeline {
public static void main(String[] args) {
Scallop scallop = new ScallopBuilder()
.addModule(createOrderFilterModule())
.addModule(createOrderProcessingModule())
.addModule(createInventoryManagementModule())
.addModule(createNotificationModule())
.build();
DataStream<Order> orderStream = new DataStreamBuilder<Order>(scallop)
.addInput("orders")
.build();
scallop.start();
orderStream.subscribe(order -> {
scallop.putData("orders", order);
});
scallop.stop();
}
private static Module createOrderFilterModule() {
return new ModuleBuilder()
.setName("orderFilterModule")
.addAction((order) -> {
return order.isUnpaid();
})
.build();
}
private static Module createOrderProcessingModule() {
return new ModuleBuilder()
.setName("orderProcessingModule")
.addAction((order) -> {
processOrder(order);
})
.build();
}
private static Module createInventoryManagementModule() {
return new ModuleBuilder()
.setName("inventoryManagementModule")
.addAction((order) -> {
updateInventory(order.getProductId());
})
.build();
}
private static Module createNotificationModule() {
return new ModuleBuilder()
.setName("notificationModule")
.addAction((order) -> {
sendNotification(order.getUserId(), order.getStatus());
})
.build();
}
private static void processOrder(Order order) {
}
private static void updateInventory(String productId) {
}
private static void sendNotification(String userId, String status) {
}
}
