001    package com.hammurapi.eventbus.tests.fastfood;
002    
003    import java.util.Collection;
004    import java.util.Collections;
005    import java.util.HashSet;
006    import java.util.concurrent.atomic.AtomicInteger;
007    
008    import com.hammurapi.eventbus.Handler;
009    import com.hammurapi.eventbus.local.LocalEventDispatchJoinContext;
010    
011    /**
012     * Handler which fulfills orders.
013     * @author Pavel Vlasov
014     *
015     */
016    public class OrderFulfiller {
017            
018            private AtomicInteger failureCounter = new AtomicInteger();
019            
020            @Handler("java(*)://order.getMainDishType().equals(mainDish.getClass())")
021            public void matchMainDish(LocalEventDispatchJoinContext<Object, Integer, Object> ctx, Order order, MainDish mainDish) {
022                    if (order.setMainDish(mainDish)) {
023                            ctx.consume(mainDish);
024                            if (order.isFulfilled()) {
025                                    ctx.consume(order);
026                                    fulfilledOrders.add(order);
027                            } else {
028                                    ctx.consumeJoin(order);
029                            }
030                    } else {
031                            failureCounter.incrementAndGet();
032                    }
033            }
034    
035            @Handler("java(*)://order.getSideDishType().equals(sideDish.getClass())")
036            public void matchSideDish(LocalEventDispatchJoinContext<Object, Integer, Object> ctx, Order order, SideDish sideDish) {
037                    if (order.setSideDish(sideDish)) {
038                            ctx.consume(sideDish);
039                            if (order.isFulfilled()) {
040                                    ctx.consume(order);
041                                    fulfilledOrders.add(order);
042                            } else {
043                                    ctx.consumeJoin(order);
044                            }
045                    } else {
046                            failureCounter.incrementAndGet();
047                    }
048            }
049            
050            public boolean isOK() {
051                    return failureCounter.get()==0;
052            }
053            
054            private Collection<Order> fulfilledOrders = Collections.synchronizedCollection(new HashSet<Order>());
055            
056            public Collection<Order> getFulfilledOrders() {
057                    return fulfilledOrders;
058            }
059    
060    }