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.ContainerReference;
  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, int port) {
  28                 super(port);
  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         deliverDeliveryPizzasContainer = useContainer(Util.DELIVER_DELIVERY_PIZZAS);
  36         }
  37
  38         public void listenForPizzas() {
  39                 SpaceListener pizzasListener = new SpaceListenerImpl(capi, preparePizzasContainer, 15000) {
  40
  41                         @Override
  42                         void onEntriesWritten(List<? extends Serializable> entries)
  43                                         throws Exception {
  44
  45                                 List<PizzaOrder> pizzas = castEntries(entries);
  46
  47                                 if(inNotification.get()) Collections.shuffle(pizzas);
  48
  49                                 for (PizzaOrder pizzaOrder : pizzas) {
  50
  51                     final boolean deliveryPizza = pizzaOrder.isDeliveryPizza();
  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                                                 // Require the lock for preparing the pizza
  57                                                 PizzaOrder order = takeMatchingEntity(new PizzaOrder(pizzaOrder.getId()), preparePizzasContainer, tx, RequestTimeout.ZERO, pizzaAlreadyCooked);
  58
  59                                                 // tell the space that you prepare the pizza -> without a transaction!!
  60                                                 Pizza pizzaInProgress = Pizza.createPizzaFromPizzaOrder(order, cookId, deliveryPizza);
  61                                                 pizzaInProgress.setStatus(PizzaOrderStatus.IN_PREPARATION);
  62
  63
  64                                                 log.info("I say that I now prepare a pizza for order {}", pizzaInProgress.getOrderId());
  65                                                 sendItemsToContainer(Arrays.asList(pizzaInProgress), pizzaInProgressContainer, RequestTimeout.DEFAULT, null);
  66
  67                                                 PizzaOrder pizza = createPizza(order);
  68
  69                         final ContainerReference container = deliveryPizza ? deliverDeliveryPizzasContainer : deliverPizzasContainer;
  70                         sendItemsToContainer(Arrays.asList(pizza), container, RequestTimeout.DEFAULT, tx);
  71
  72                                                 capi.commitTransaction(tx);
  73                                                 log.info("I have completed preparing a pizza for order {}!", pizza.getOrderId());
  74                                         } catch (NullPointerException e) {
  75                                                 // the strange nullpointer exception from the space
  76                                         } catch (Exception e) {
  77                                                 log.info(e.getMessage());
  78                                         }
  79
  80                                 }
  81             }
  82
  83                         @Override
  84                         List<? extends Serializable> getEntries() throws Exception {
  85                                 return capi.read(preparePizzasContainer, FifoCoordinator
  86                                                 .newSelector(MzsConstants.Selecting.COUNT_ALL),
  87                                                 RequestTimeout.DEFAULT, null);
  88                         }
  89         };
  90         createNotification(pizzasListener, preparePizzasContainer);
  91
  92         }
  93
  94         private PizzaOrder createPizza(PizzaOrder order) throws InterruptedException {
  95                 long duration = order.getPizzaType().duration;
  96                 Thread.sleep(duration * 1000);
  97
  98                 PizzaOrder pizza = Pizza.createPizzaFromPizzaOrder(order, cookId, false);
  99                 pizza.setStatus(PizzaOrderStatus.DONE);
 100                 return pizza;
 101         }
 102
 103 }