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