src/main/java/at/ac/tuwien/sbc/valesriegler/xvsm/CookXVSM.java

   1 package at.ac.tuwien.sbc.valesriegler.xvsm;
   2
   3 import java.io.Serializable;
   4 import java.net.URI;
   5 import java.util.Arrays;
   6 import java.util.Collections;
   7 import java.util.List;
   8
   9 import at.ac.tuwien.sbc.valesriegler.types.*;
  10 import org.mozartspaces.capi3.FifoCoordinator;
  11 import org.mozartspaces.core.ContainerReference;
  12 import org.mozartspaces.core.MzsConstants;
  13 import org.mozartspaces.core.MzsConstants.RequestTimeout;
  14 import org.mozartspaces.core.MzsCoreException;
  15 import org.mozartspaces.core.TransactionReference;
  16 import org.slf4j.Logger;
  17 import org.slf4j.LoggerFactory;
  18
  19 import at.ac.tuwien.sbc.valesriegler.common.Util;
  20
  21 public class CookXVSM extends AbstractXVSMConnector {
  22         private static final Logger log = LoggerFactory.getLogger(CookXVSM.class);
  23
  24         private int cookId;
  25
  26         public CookXVSM(int id, int port) {
  27                 super(port);
  28
  29                 this.cookId = id;
  30                 orderTakenContainer = useContainer(Util.ORDER_TAKEN);
  31                 deliveryOrderTakenContainer = useContainer(Util.DELIVERY_ORDER_TAKEN);
  32                 preparedPizzasContainer = useContainer(Util.DELIVER_PIZZAS);
  33                 preparePizzasContainer = useContainer(Util.PREPARE_PIZZAS);
  34                 prepareDeliveryPizzasContainer = useContainer(Util.PREPARE_DELIVERY_PIZZAS);
  35                 pizzaInProgressContainer = useContainer(Util.PIZZAS_IN_PROGRESS) ;
  36         preparedDeliveryPizzasContainer = useContainer(Util.DELIVER_DELIVERY_PIZZAS);
  37         }
  38
  39     public void listenForPizzas() {
  40         SpaceListener pizzasListener = new SpaceListenerImpl(capi, preparePizzasContainer, 15000) {
  41
  42             @Override
  43             void onEntriesWritten(List<? extends Serializable> entries)
  44                     throws Exception {
  45
  46                 List<PizzaOrder> pizzas = castEntries(entries);
  47
  48                 if(inNotification.get()) Collections.shuffle(pizzas);
  49
  50                 for (PizzaOrder pizzaOrder : pizzas) {
  51
  52                     TransactionReference tx = capi.createTransaction(9000, URI.create(String.format(Util.SERVER_ADDR, port)));
  53                     String pizzaAlreadyCooked = String.format("Pizza with id %d has already been cooked by another cook", pizzaOrder.getId());
  54
  55                     try {
  56                         if (! mayPreparePizza(pizzaOrder, tx)) {
  57                             capi.rollbackTransaction(tx);
  58                             continue;
  59                         }
  60
  61                         // Require the lock for preparing the pizza
  62                         PizzaOrder order = takeMatchingEntity(new PizzaOrder(pizzaOrder.getId()), preparePizzasContainer, tx, RequestTimeout.ZERO, pizzaAlreadyCooked);
  63
  64
  65                         // tell the space that you prepare the pizza -> without a transaction!!
  66                         Pizza pizzaInProgress = Pizza.createPizzaFromPizzaOrder(order, cookId, true);
  67                         pizzaInProgress.setStatus(PizzaOrderStatus.IN_PREPARATION);
  68
  69
  70                         log.info("I say that I now prepare a pizza for order {}", pizzaInProgress.getOrderId());
  71                         sendItemsToContainer(Arrays.asList(pizzaInProgress), pizzaInProgressContainer, RequestTimeout.DEFAULT, null);
  72
  73                         PizzaOrder pizza = createPizza(order);
  74
  75                         final ContainerReference container = preparedPizzasContainer;
  76                         sendItemsToContainer(Arrays.asList(pizza), container, RequestTimeout.DEFAULT, tx);
  77
  78                         capi.commitTransaction(tx);
  79                         log.info("I have completed preparing a pizza for order {}!", pizza.getOrderId());
  80                     } catch (NullPointerException e) {
  81                         // the strange nullpointer exception from the space
  82                     } catch (Exception e) {
  83                         log.info(e.getMessage());
  84                     }
  85
  86                 }
  87             }
  88
  89             @Override
  90             List<? extends Serializable> getEntries() throws Exception {
  91                 return capi.read(preparePizzasContainer, FifoCoordinator
  92                         .newSelector(MzsConstants.Selecting.COUNT_ALL),
  93                         RequestTimeout.DEFAULT, null);
  94             }
  95         };
  96         createNotification(pizzasListener, preparePizzasContainer);
  97
  98     }
  99
 100     /**
 101      * A cook may prepare a pizza if there is no current delivery order or if at least a pizza of the same order is
 102      * in preparation!
 103      */
 104     private boolean mayPreparePizza(PizzaOrder pizzaOrder, TransactionReference tx) throws MzsCoreException {
 105         List<PizzaOrder> deliveryPizzas = readMatchingEntities(new PizzaOrder(), prepareDeliveryPizzasContainer, tx, RequestTimeout.ZERO, "BBBBBB");
 106         if (! deliveryPizzas.isEmpty()) {
 107             final PizzaOrder template = new PizzaOrder();
 108             template.setOrderId(pizzaOrder.getOrderId());
 109             final List<PizzaOrder> pizzasOfOrderInProgress = readMatchingEntities(template, pizzaInProgressContainer, tx, RequestTimeout.ZERO, "ASDF");
 110             if(pizzasOfOrderInProgress.isEmpty()) return false;
 111         }
 112         return true;
 113     }
 114
 115
 116     public void listenForDeliveryPizzas() {
 117         SpaceListener pizzasListener = new SpaceListenerImpl(capi, prepareDeliveryPizzasContainer, 15000) {
 118
 119             @Override
 120             void onEntriesWritten(List<? extends Serializable> entries)
 121                     throws Exception {
 122
 123                 List<PizzaOrder> pizzas = castEntries(entries);
 124
 125                 if(inNotification.get()) Collections.shuffle(pizzas);
 126
 127                 for (PizzaOrder pizzaOrder : pizzas) {
 128
 129                     TransactionReference tx = capi.createTransaction(9000, URI.create(String.format(Util.SERVER_ADDR, port)));
 130                     String pizzaAlreadyCooked = String.format("Pizza with id %d has already been cooked by another cook", pizzaOrder.getId());
 131
 132                     try {
 133                         // Require the lock for preparing the pizza
 134                         PizzaOrder order = takeMatchingEntity(new PizzaOrder(pizzaOrder.getId()), prepareDeliveryPizzasContainer, tx, RequestTimeout.ZERO, pizzaAlreadyCooked);
 135
 136                         // tell the space that you prepare the pizza -> without a transaction!!
 137                         Pizza pizzaInProgress = Pizza.createPizzaFromPizzaOrder(order, cookId, true);
 138                         pizzaInProgress.setStatus(PizzaOrderStatus.IN_PREPARATION);
 139
 140
 141                         log.info("I say that I now prepare a pizza for order {}", pizzaInProgress.getOrderId());
 142                         sendItemsToContainer(Arrays.asList(pizzaInProgress), pizzaInProgressContainer, RequestTimeout.DEFAULT, null);
 143
 144                         PizzaOrder pizza = createPizza(order);
 145
 146                         final ContainerReference container = preparedDeliveryPizzasContainer;
 147                         sendItemsToContainer(Arrays.asList(pizza), container, RequestTimeout.DEFAULT, tx);
 148
 149                         capi.commitTransaction(tx);
 150                         log.info("I have completed preparing a delivery pizza for order {}!", pizza.getOrderId());
 151                     } catch (NullPointerException e) {
 152                         // the strange nullpointer exception from the space
 153                     } catch (Exception e) {
 154                         log.info(e.getMessage());
 155                     }
 156
 157                 }
 158             }
 159
 160             @Override
 161             List<? extends Serializable> getEntries() throws Exception {
 162                 return capi.read(prepareDeliveryPizzasContainer, FifoCoordinator
 163                         .newSelector(MzsConstants.Selecting.COUNT_ALL),
 164                         RequestTimeout.DEFAULT, null);
 165             }
 166         };
 167         createNotification(pizzasListener, prepareDeliveryPizzasContainer);
 168
 169     }
 170
 171         private PizzaOrder createPizza(PizzaOrder order) throws InterruptedException {
 172                 long duration = order.getPizzaType().duration;
 173                 Thread.sleep(duration * 1000);
 174
 175                 PizzaOrder pizza = Pizza.createPizzaFromPizzaOrder(order, cookId, false);
 176                 pizza.setStatus(PizzaOrderStatus.DONE);
 177                 return pizza;
 178         }
 179
 180 }