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 org.mozartspaces.capi3.FifoCoordinator;
  10 import org.mozartspaces.capi3.LindaCoordinator;
  11 import org.mozartspaces.core.MzsConstants;
  12 import org.mozartspaces.core.MzsConstants.RequestTimeout;
  13 import org.mozartspaces.core.TransactionReference;
  14 import org.slf4j.Logger;
  15 import org.slf4j.LoggerFactory;
  16
  17 import at.ac.tuwien.sbc.valesriegler.common.Util;
  18 import at.ac.tuwien.sbc.valesriegler.types.Pizza;
  19 import at.ac.tuwien.sbc.valesriegler.types.PizzaOrder;
  20 import at.ac.tuwien.sbc.valesriegler.types.PizzaOrderStatus;
  21
  22 public class CookXVSM extends AbstractXVSMConnector {
  23         private static final Logger log = LoggerFactory.getLogger(CookXVSM.class);
  24
  25         private int cookId;
  26
  27         public CookXVSM(int id) {
  28                 super();
  29
  30                 this.cookId = id;
  31                 orderTakenContainer = useContainer(Util.ORDER_TAKEN) ;
  32                 deliverPizzasContainer = useContainer(Util.DELIVER_PIZZAS);
  33                 preparePizzasContainer = useContainer(Util.PREPARE_PIZZAS);
  34                 pizzaInProgressContainer = useContainer(Util.PIZZAS_IN_PROGRESS) ;
  35         }
  36
  37         public void listenForPizzas() {
  38                 SpaceListener pizzasListener = new SpaceListenerImpl(capi, preparePizzasContainer, 15000) {
  39
  40                         @Override
  41                         void onEntriesWritten(List<? extends Serializable> entries)
  42                                         throws Exception {
  43
  44                                 List<PizzaOrder> pizzas = castEntries(entries);
  45
  46                                 if(inNotification.get()) Collections.shuffle(pizzas);
  47
  48                                 for (PizzaOrder pizzaOrder : pizzas) {
  49
  50                                         TransactionReference tx = capi.createTransaction(9000, URI.create(Util.SERVER_ADDR));
  51                                         String pizzaAlreadyCooked = String.format("Pizza with id %d has already been cooked by another cook", pizzaOrder.getId());
  52
  53                                         try {
  54                                                 // Require the lock for preparing the pizza
  55                                                 PizzaOrder order = takeMatchingEntity(new PizzaOrder(pizzaOrder.getId()), preparePizzasContainer, tx, RequestTimeout.ZERO, pizzaAlreadyCooked);
  56
  57                                                 // tell the space that you prepare the pizza -> without a transaction!!
  58                                                 Pizza pizzaInProgress = Pizza.createPizzaFromPizzaOrder(order, cookId);
  59                                                 pizzaInProgress.setStatus(PizzaOrderStatus.IN_PREPARATION);
  60
  61
  62                                                 log.info("I say that I now prepare a pizza for order {}", pizzaInProgress.getOrderId());
  63                                                 sendItemsToContainer(Arrays.asList(pizzaInProgress), pizzaInProgressContainer, RequestTimeout.DEFAULT, null);
  64
  65                                                 PizzaOrder pizza = createPizza(order);
  66
  67                                                 sendItemsToContainer(Arrays.asList(pizza), deliverPizzasContainer, RequestTimeout.DEFAULT, tx);
  68
  69                                                 capi.commitTransaction(tx);
  70                                                 log.info("I have completed preparing a pizza for order {}!", pizza.getOrderId());
  71                                         } catch (NullPointerException e) {
  72                                                 // the strange nullpointer exception from the space
  73                                         } catch (Exception e) {
  74                                                 log.info(e.getMessage());
  75                                         }
  76
  77                                 }
  78             }
  79
  80                         @Override
  81                         List<? extends Serializable> getEntries() throws Exception {
  82                                 return capi.read(preparePizzasContainer, FifoCoordinator
  83                                                 .newSelector(MzsConstants.Selecting.COUNT_ALL),
  84                                                 RequestTimeout.DEFAULT, null);
  85                         }
  86         };
  87         createNotification(pizzasListener, preparePizzasContainer);
  88
  89         }
  90
  91         private PizzaOrder createPizza(PizzaOrder order) throws InterruptedException {
  92                 long duration = order.getPizzaType().duration;
  93                 Thread.sleep(duration * 1000);
  94
  95                 PizzaOrder pizza = Pizza.createPizzaFromPizzaOrder(order, cookId);
  96                 pizza.setStatus(PizzaOrderStatus.DONE);
  97                 return pizza;
  98         }
  99
 100 }