// Path: ch05/nile-carts/src/main/java/nile/events/AbandonedCartEvent.java
// public class AbandonedCartEvent extends Event {
//   public final DirectObject directObject;
// 
//   public AbandonedCartEvent(String shopper, String cart) {
//     super(shopper, "abandon");
//     this.directObject = new DirectObject(cart);
//   }
// 
//   public static final class DirectObject {
//     public final Cart cart;
// 
//     public DirectObject(String cart) {
//       this.cart = new Cart(cart);
//     }
// 
//     public static final class Cart {
// 
//       private static final int ABANDONED_AFTER_SECS = 1800;            // a
// 
//       public List<Map<String, Object>> items =
//         new ArrayList<Map<String, Object>>();
// 
//       public Cart(String json) {
//         if (json != null) {
//           try {
//             this.items = MAPPER.readValue(json,
//               new TypeReference<List<Map<String, Object>>>() {});
//           } catch (IOException ioe) {
//             throw new RuntimeException("Problem parsing JSON cart", ioe);
//           }
//         }
//       }
// 
//       public void addItem(Map<String, Object> item) {                  // b
//         this.items.add(item);
//       }
// 
//       public String asJson() {                                         // c
//         try {
//           return MAPPER.writeValueAsString(this.items);
//         } catch (IOException ioe) {
//           throw new RuntimeException("Problem writing JSON cart", ioe);
//         }
//       }
// 
//       public static boolean isAbandoned(String timestamp) {            // d
//         DateTime ts = EVENT_DTF.parseDateTime(timestamp);
//         DateTime cutoff = new DateTime(DateTimeZone.UTC)
//           .minusSeconds(ABANDONED_AFTER_SECS);
//         return ts.isBefore(cutoff);
//       }
//     }
//   }
// }
// 
// Path: ch05/nile-carts/src/main/java/nile/events/AbandonedCartEvent.java
// public static final class Cart {
// 
//   private static final int ABANDONED_AFTER_SECS = 1800;            // a
// 
//   public List<Map<String, Object>> items =
//     new ArrayList<Map<String, Object>>();
// 
//   public Cart(String json) {
//     if (json != null) {
//       try {
//         this.items = MAPPER.readValue(json,
//           new TypeReference<List<Map<String, Object>>>() {});
//       } catch (IOException ioe) {
//         throw new RuntimeException("Problem parsing JSON cart", ioe);
//       }
//     }
//   }
// 
//   public void addItem(Map<String, Object> item) {                  // b
//     this.items.add(item);
//   }
// 
//   public String asJson() {                                         // c
//     try {
//       return MAPPER.writeValueAsString(this.items);
//     } catch (IOException ioe) {
//       throw new RuntimeException("Problem writing JSON cart", ioe);
//     }
//   }
// 
//   public static boolean isAbandoned(String timestamp) {            // d
//     DateTime ts = EVENT_DTF.parseDateTime(timestamp);
//     DateTime cutoff = new DateTime(DateTimeZone.UTC)
//       .minusSeconds(ABANDONED_AFTER_SECS);
//     return ts.isBefore(cutoff);
//   }
// }


import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import org.apache.samza.config.Config;
import org.apache.samza.storage.kv.KeyValueStore;
import org.apache.samza.storage.kv.KeyValueIterator;
import org.apache.samza.storage.kv.Entry;
import org.apache.samza.system.IncomingMessageEnvelope;
import org.apache.samza.system.OutgoingMessageEnvelope;
import org.apache.samza.system.SystemStream;
import org.apache.samza.task.InitableTask;
import org.apache.samza.task.MessageCollector;
import org.apache.samza.task.StreamTask;
import org.apache.samza.task.TaskContext;
import org.apache.samza.task.TaskCoordinator;
import org.apache.samza.task.WindowableTask;
import nile.events.AbandonedCartEvent;
import nile.events.AbandonedCartEvent.DirectObject.Cart;
    if (verb.equals("add")) {                                          // a
      String timestamp = (String) ((Map<String, Object>)
        event.get("context")).get("timestamp");

      Map<String, Object> item = (Map<String, Object>)
        ((Map<String, Object>) event.get("directObject")).get("item");
      Cart cart = new Cart(store.get(asCartKey(shopper)));
      cart.addItem(item);

      store.put(asTimestampKey(shopper), timestamp);
      store.put(asCartKey(shopper), cart.asJson());
    
    } else if (verb.equals("place")) {                                 // b
      resetShopper(shopper);
    }
  }

  @Override
  public void window(MessageCollector collector,
    TaskCoordinator coordinator) {

    KeyValueIterator<String, String> entries = store.all();
    while (entries.hasNext()) {                                        // c
      Entry<String, String> entry = entries.next();
      String key = entry.getKey();
      String value = entry.getValue();
      if (isTimestampKey(key) && Cart.isAbandoned(value)) {            // d
        String shopper = extractShopper(key);
        String cart = store.get(asCartKey(shopper));
        
// Path: ch05/nile-carts/src/main/java/nile/events/AbandonedCartEvent.java
// public class AbandonedCartEvent extends Event {
//   public final DirectObject directObject;
// 
//   public AbandonedCartEvent(String shopper, String cart) {
//     super(shopper, "abandon");
//     this.directObject = new DirectObject(cart);
//   }
// 
//   public static final class DirectObject {
//     public final Cart cart;
// 
//     public DirectObject(String cart) {
//       this.cart = new Cart(cart);
//     }
// 
//     public static final class Cart {
// 
//       private static final int ABANDONED_AFTER_SECS = 1800;            // a
// 
//       public List<Map<String, Object>> items =
//         new ArrayList<Map<String, Object>>();
// 
//       public Cart(String json) {
//         if (json != null) {
//           try {
//             this.items = MAPPER.readValue(json,
//               new TypeReference<List<Map<String, Object>>>() {});
//           } catch (IOException ioe) {
//             throw new RuntimeException("Problem parsing JSON cart", ioe);
//           }
//         }
//       }
// 
//       public void addItem(Map<String, Object> item) {                  // b
//         this.items.add(item);
//       }
// 
//       public String asJson() {                                         // c
//         try {
//           return MAPPER.writeValueAsString(this.items);
//         } catch (IOException ioe) {
//           throw new RuntimeException("Problem writing JSON cart", ioe);
//         }
//       }
// 
//       public static boolean isAbandoned(String timestamp) {            // d
//         DateTime ts = EVENT_DTF.parseDateTime(timestamp);
//         DateTime cutoff = new DateTime(DateTimeZone.UTC)
//           .minusSeconds(ABANDONED_AFTER_SECS);
//         return ts.isBefore(cutoff);
//       }
//     }
//   }
// }
// 
// Path: ch05/nile-carts/src/main/java/nile/events/AbandonedCartEvent.java
// public static final class Cart {
// 
//   private static final int ABANDONED_AFTER_SECS = 1800;            // a
// 
//   public List<Map<String, Object>> items =
//     new ArrayList<Map<String, Object>>();
// 
//   public Cart(String json) {
//     if (json != null) {
//       try {
//         this.items = MAPPER.readValue(json,
//           new TypeReference<List<Map<String, Object>>>() {});
//       } catch (IOException ioe) {
//         throw new RuntimeException("Problem parsing JSON cart", ioe);
//       }
//     }
//   }
// 
//   public void addItem(Map<String, Object> item) {                  // b
//     this.items.add(item);
//   }
// 
//   public String asJson() {                                         // c
//     try {
//       return MAPPER.writeValueAsString(this.items);
//     } catch (IOException ioe) {
//       throw new RuntimeException("Problem writing JSON cart", ioe);
//     }
//   }
// 
//   public static boolean isAbandoned(String timestamp) {            // d
//     DateTime ts = EVENT_DTF.parseDateTime(timestamp);
//     DateTime cutoff = new DateTime(DateTimeZone.UTC)
//       .minusSeconds(ABANDONED_AFTER_SECS);
//     return ts.isBefore(cutoff);
//   }
// }


// Path: ch05/nile-carts/src/main/java/nile/tasks/AbandonedCartStreamTask.java
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import org.apache.samza.config.Config;
import org.apache.samza.storage.kv.KeyValueStore;
import org.apache.samza.storage.kv.KeyValueIterator;
import org.apache.samza.storage.kv.Entry;
import org.apache.samza.system.IncomingMessageEnvelope;
import org.apache.samza.system.OutgoingMessageEnvelope;
import org.apache.samza.system.SystemStream;
import org.apache.samza.task.InitableTask;
import org.apache.samza.task.MessageCollector;
import org.apache.samza.task.StreamTask;
import org.apache.samza.task.TaskContext;
import org.apache.samza.task.TaskCoordinator;
import org.apache.samza.task.WindowableTask;
import nile.events.AbandonedCartEvent;
import nile.events.AbandonedCartEvent.DirectObject.Cart;

    if (verb.equals("add")) {                                          // a
      String timestamp = (String) ((Map<String, Object>)
        event.get("context")).get("timestamp");

      Map<String, Object> item = (Map<String, Object>)
        ((Map<String, Object>) event.get("directObject")).get("item");
      Cart cart = new Cart(store.get(asCartKey(shopper)));
      cart.addItem(item);

      store.put(asTimestampKey(shopper), timestamp);
      store.put(asCartKey(shopper), cart.asJson());
    
    } else if (verb.equals("place")) {                                 // b
      resetShopper(shopper);
    }
  }

  @Override
  public void window(MessageCollector collector,
    TaskCoordinator coordinator) {

    KeyValueIterator<String, String> entries = store.all();
    while (entries.hasNext()) {                                        // c
      Entry<String, String> entry = entries.next();
      String key = entry.getKey();
      String value = entry.getValue();
      if (isTimestampKey(key) && Cart.isAbandoned(value)) {            // d
        String shopper = extractShopper(key);
        String cart = store.get(asCartKey(shopper));
        
        AbandonedCartEvent event =