> processCargoRegistered(CargoRegistered event, UUID offset) {
+ BoundStatement bindWriteCargo = writeCargo.bind();
+ bindWriteCargo.setString("cargoId", event.getCargo().getId());
+ bindWriteCargo.setString("name", event.getCargo().getName());
+ bindWriteCargo.setString("description", event.getCargo().getDescription());
+ bindWriteCargo.setString("owner", event.getCargo().getOwner());
+ bindWriteCargo.setString("destination", event.getCargo().getDestination());
+ BoundStatement bindWriteOffset = writeOffset.bind(offset);
+ log.info("Persisted {}", event.getCargo().getId());
+ return completedStatements(Arrays.asList(bindWriteCargo, bindWriteOffset));
+ }
+
+}
diff --git a/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationCommand.java b/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationCommand.java
new file mode 100644
index 000000000000..8abdf955f61d
--- /dev/null
+++ b/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationCommand.java
@@ -0,0 +1,36 @@
+package sample.cargotracker.registration.impl;
+
+import akka.Done;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.lightbend.lagom.javadsl.immutable.ImmutableStyle;
+import com.lightbend.lagom.javadsl.persistence.PersistentEntity;
+import com.lightbend.lagom.serialization.CompressedJsonable;
+import com.lightbend.lagom.serialization.Jsonable;
+import org.immutables.value.Value;
+import sample.cargotracker.registration.api.Cargo;
+
+/**
+ * This interface defines all the commands that the Cargo entity supports.
+ *
+ * By convention, the commands should be inner classes of the interface, which makes it simple to get a
+ * complete picture of what commands an entity supports.
+ */
+public interface RegistrationCommand extends Jsonable {
+
+ /**
+ * A command to register cargo.
+ *
+ * It has a reply type of {@link akka.Done}, which is sent back to the caller when all the events
+ * emitted by this command are successfully persisted.
+ */
+ @Value.Immutable
+ @ImmutableStyle
+ @JsonDeserialize(as = RegisterCargo.class)
+ public interface AbstractRegisterCargo extends RegistrationCommand, CompressedJsonable,
+ PersistentEntity.ReplyType {
+
+ @Value.Parameter
+ Cargo getCargo();
+ }
+
+}
diff --git a/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationEvent.java b/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationEvent.java
new file mode 100644
index 000000000000..f0c01d4eabad
--- /dev/null
+++ b/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationEvent.java
@@ -0,0 +1,40 @@
+package sample.cargotracker.registration.impl;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.lightbend.lagom.javadsl.immutable.ImmutableStyle;
+import com.lightbend.lagom.javadsl.persistence.AggregateEvent;
+import com.lightbend.lagom.javadsl.persistence.AggregateEventTag;
+import com.lightbend.lagom.serialization.Jsonable;
+import org.immutables.value.Value;
+import org.immutables.value.Value.Immutable;
+import sample.cargotracker.registration.api.Cargo;
+
+/**
+ * This interface defines all the events that the Cargo entity supports.
+ *
+ * By convention, the events should be inner classes of the interface, which makes it simple to get a
+ * complete picture of what events an entity has.
+ */
+public interface RegistrationEvent extends Jsonable , AggregateEvent {
+
+ /**
+ * An event that represents a new cargo registration .
+ */
+ @Immutable
+ @ImmutableStyle
+ @JsonDeserialize(as = CargoRegistered.class)
+ interface AbstractCargoRegistered extends RegistrationEvent {
+
+ @Override
+ default public AggregateEventTag aggregateTag() {
+ return RegistrationEventTag.INSTANCE;
+ }
+
+
+ @Value.Parameter
+ String getId();
+
+ @Value.Parameter
+ Cargo getCargo();
+ }
+}
diff --git a/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationEventTag.java b/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationEventTag.java
new file mode 100644
index 000000000000..85504cc9c911
--- /dev/null
+++ b/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationEventTag.java
@@ -0,0 +1,13 @@
+package sample.cargotracker.registration.impl;
+
+import com.lightbend.lagom.javadsl.persistence.AggregateEventTag;
+
+/**
+ * Register a common event tag
+ */
+public class RegistrationEventTag {
+
+ public static final AggregateEventTag INSTANCE =
+ AggregateEventTag.of(RegistrationEvent.class);
+
+}
diff --git a/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationServiceImpl.java b/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationServiceImpl.java
new file mode 100644
index 000000000000..77cb0db9fa82
--- /dev/null
+++ b/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationServiceImpl.java
@@ -0,0 +1,138 @@
+package sample.cargotracker.registration.impl;
+
+import akka.Done;
+import akka.NotUsed;
+import akka.stream.javadsl.Source;
+import com.google.inject.Inject;
+import com.lightbend.lagom.javadsl.api.ServiceCall;
+import com.lightbend.lagom.javadsl.persistence.PersistentEntityRef;
+import com.lightbend.lagom.javadsl.persistence.PersistentEntityRegistry;
+import com.lightbend.lagom.javadsl.persistence.cassandra.CassandraReadSide;
+import com.lightbend.lagom.javadsl.persistence.cassandra.CassandraSession;
+import com.lightbend.lagom.javadsl.pubsub.PubSubRef;
+import com.lightbend.lagom.javadsl.pubsub.PubSubRegistry;
+import com.lightbend.lagom.javadsl.pubsub.TopicId;
+import org.pcollections.PSequence;
+import org.pcollections.TreePVector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sample.cargotracker.registration.api.Cargo;
+import sample.cargotracker.registration.api.RegistrationService;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.stream.Collectors;
+
+/**
+ * Implementation of the RegistrationService.
+ * Look at circuit breaker information: http://localhost:25003/_status/circuit-breaker/current
+ */
+public class RegistrationServiceImpl implements RegistrationService {
+
+ private final PersistentEntityRegistry persistentEntityRegistry;
+ private final PubSubRegistry topics;
+ private final CassandraSession db;
+ private final Logger log = LoggerFactory.getLogger(RegistrationServiceImpl.class);
+
+ /**
+ * Constructor with the relevant infrastructure elements injected.
+ *
+ * @param topics
+ * @param persistentEntityRegistry
+ * @param readSide
+ * @param db
+ */
+ @Inject
+ public RegistrationServiceImpl(PubSubRegistry topics, PersistentEntityRegistry persistentEntityRegistry, CassandraReadSide readSide,
+ CassandraSession db) {
+ this.persistentEntityRegistry = persistentEntityRegistry;
+ this.topics = topics;
+ this.db = db;
+ persistentEntityRegistry.register(CargoEntity.class);
+ readSide.register(CargoEventProcessor.class);
+ }
+
+ /**
+ * Register Cargo service call
+ *
+ * @return
+ */
+ @Override
+ public ServiceCall register() {
+ return (id, request) -> {
+ /* Publish received entity into topic named "Topic" */
+ PubSubRef topic = topics.refFor(TopicId.of(Cargo.class, "topic"));
+ topic.publish(request);
+ log.info("Cargo ID: {}.", request.getId());
+ /* Look up the Cargo entity for the given ID. */
+ PersistentEntityRef ref =
+ persistentEntityRegistry.refFor(CargoEntity.class, request.getId());
+ /* Tell the entity to use the Cargo information in the request. */
+ return ref.ask(RegisterCargo.of(request));
+ };
+ }
+
+ /**
+ * Get live registrations service call
+ *
+ * @return
+ */
+ @Override
+ public ServiceCall> getLiveRegistrations() {
+ return (id, req) -> {
+ PubSubRef topic = topics.refFor(TopicId.of(Cargo.class, "topic"));
+ return CompletableFuture.completedFuture(topic.subscriber());
+ };
+ }
+
+
+ /**
+ * Get all persisted Cargo services call
+ * Websockets capable
+ *
+ * @return
+ * @deprecated
+ */
+ public ServiceCall> getAllRegistrationsOld() {
+ log.info("Select all cargo .");
+ return (id, req) -> {
+ Source result = db.select(
+ "SELECT cargoId, name, description, owner, destination FROM cargo;").map(row ->
+ Cargo.of(row.getString("cargoId"),
+ row.getString("name"),
+ row.getString("description"),
+ row.getString("owner"),
+ row.getString("destination")));
+ return CompletableFuture.completedFuture(result);
+
+ };
+ }
+
+ /**
+ * Get all registered Cargo
+ *
+ * @return
+ */
+ @Override
+ public ServiceCall> getAllRegistrations() {
+ return (userId, req) -> {
+ CompletionStage> result = db.selectAll("SELECT cargoid, name, description, owner, destination FROM cargo")
+ .thenApply(rows -> {
+ List cargos = rows.stream().map(row -> Cargo.of(row.getString("cargoid"),
+ row.getString("name"),
+ row.getString("description"),
+ row.getString("owner"),
+ row.getString("destination"))).collect(Collectors.toList());
+ return TreePVector.from(cargos);
+ });
+ return result;
+ };
+ }
+
+ public ServiceCall getRegistration() {
+
+ //TODO Implement meaningful
+ return null;
+ }
+}
diff --git a/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationServiceModule.java b/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationServiceModule.java
new file mode 100644
index 000000000000..bebc97db4cdd
--- /dev/null
+++ b/lagom-cargotracker/registration-impl/src/main/java/sample/cargotracker/registration/impl/RegistrationServiceModule.java
@@ -0,0 +1,15 @@
+package sample.cargotracker.registration.impl;
+
+import com.google.inject.AbstractModule;
+import com.lightbend.lagom.javadsl.server.ServiceGuiceSupport;
+import sample.cargotracker.registration.api.RegistrationService;
+
+/**
+ * The module that binds the RegistrationService so that it can be served.
+ */
+public class RegistrationServiceModule extends AbstractModule implements ServiceGuiceSupport {
+ @Override
+ protected void configure() {
+ bindServices(serviceBinding(RegistrationService.class, RegistrationServiceImpl.class));
+ }
+}
diff --git a/lagom-cargotracker/registration-impl/src/main/resources/application.conf b/lagom-cargotracker/registration-impl/src/main/resources/application.conf
new file mode 100644
index 000000000000..96cfb8673531
--- /dev/null
+++ b/lagom-cargotracker/registration-impl/src/main/resources/application.conf
@@ -0,0 +1 @@
+play.modules.enabled += sample.cargotracker.registration.impl.RegistrationServiceModule
diff --git a/lagom-cargotracker/registration-impl/src/test/java/RegistrationServiceTest.java b/lagom-cargotracker/registration-impl/src/test/java/RegistrationServiceTest.java
new file mode 100644
index 000000000000..90b6b0340fb4
--- /dev/null
+++ b/lagom-cargotracker/registration-impl/src/test/java/RegistrationServiceTest.java
@@ -0,0 +1,22 @@
+
+import static com.lightbend.lagom.javadsl.testkit.ServiceTest.*;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.Assert.assertEquals;
+import org.junit.Test;
+import sample.cargotracker.registration.api.RegistrationService;
+
+/**
+ * Created by myfear on 03/04/16.
+ */
+public class RegistrationServiceTest {
+
+ @Test
+ public void shouldSayHello() throws Exception {
+ withServer(defaultSetup(), server -> {
+ RegistrationService service = server.client(RegistrationService.class);
+
+ //TODO add meaningfull test
+ });
+ }
+
+}
diff --git a/lagom-cargotracker/shipping-api/src/main/java/sample/cargotracker/shipping/ErrorHandler.java b/lagom-cargotracker/shipping-api/src/main/java/sample/cargotracker/shipping/ErrorHandler.java
new file mode 100644
index 000000000000..6a3bcfe4cce4
--- /dev/null
+++ b/lagom-cargotracker/shipping-api/src/main/java/sample/cargotracker/shipping/ErrorHandler.java
@@ -0,0 +1,69 @@
+package sample.cargotracker.shipping;
+
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import play.Configuration;
+import play.Environment;
+import play.api.OptionalSourceMapper;
+import play.api.UsefulException;
+import play.api.routing.Router;
+import play.http.DefaultHttpErrorHandler;
+import play.libs.Json;
+import play.mvc.Http;
+import play.mvc.Result;
+import play.mvc.Results;
+
+import javax.inject.Inject;
+import javax.inject.Provider;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+
+/**
+ * Overrides the default HTTP error handler to return JSON
+ */
+public class ErrorHandler extends DefaultHttpErrorHandler {
+
+ @Inject
+ public ErrorHandler(Configuration configuration, Environment environment, OptionalSourceMapper sourceMapper, Provider routes) {
+ super(configuration, environment, sourceMapper, routes);
+ }
+
+ /**
+ * Invoked in dev mode when a server error occurs.
+ *
+ * @param request The request that triggered the error.
+ * @param exception The exception.
+ */
+ protected CompletionStage onDevServerError(Http.RequestHeader request, UsefulException exception) {
+
+ ObjectNode jsonError = Json.newObject();
+
+ final Throwable cause = exception.cause;
+ final String description = exception.description;
+ final String id = exception.id;
+ final String title = exception.title;
+
+ jsonError.put("description", description);
+ jsonError.put("title", title);
+ jsonError.put("id", id);
+ jsonError.put("message", exception.getMessage());
+ jsonError.set("cause", causesToJson(cause));
+
+ return CompletableFuture.completedFuture(Results.internalServerError(jsonError));
+ }
+
+ private ArrayNode causesToJson(Throwable throwable) {
+ ArrayNode causesNode = JsonNodeFactory.instance.arrayNode();
+
+ while (throwable != null) {
+ ObjectNode causeNode = causesNode.addObject();
+ causeNode.put("message", throwable.getMessage());
+ causeNode.put("type", throwable.getClass().getName());
+
+ throwable = throwable.getCause();
+ }
+
+ return causesNode;
+ }
+}
diff --git a/lagom-cargotracker/shipping-api/src/main/java/sample/cargotracker/shipping/api/AbstractItinerary.java b/lagom-cargotracker/shipping-api/src/main/java/sample/cargotracker/shipping/api/AbstractItinerary.java
new file mode 100644
index 000000000000..17bd377bc935
--- /dev/null
+++ b/lagom-cargotracker/shipping-api/src/main/java/sample/cargotracker/shipping/api/AbstractItinerary.java
@@ -0,0 +1,32 @@
+package sample.cargotracker.shipping.api;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.lightbend.lagom.javadsl.immutable.ImmutableStyle;
+
+import java.util.List;
+
+import org.immutables.value.Value;
+
+import org.pcollections.PSequence;
+import sample.cargotracker.shipping.api.Leg;
+
+@Value.Immutable
+@ImmutableStyle
+@JsonDeserialize(as = Itinerary.class)
+public interface AbstractItinerary {
+
+ @Value.Parameter
+ String getId();
+
+ @Value.Parameter
+ String getCargoId();
+
+ @Value.Parameter
+ String getOrigin();
+
+ @Value.Parameter
+ String getDestination();
+
+ @Value.Parameter
+ PSequence getLegs();
+}
diff --git a/lagom-cargotracker/shipping-api/src/main/java/sample/cargotracker/shipping/api/AbstractLeg.java b/lagom-cargotracker/shipping-api/src/main/java/sample/cargotracker/shipping/api/AbstractLeg.java
new file mode 100644
index 000000000000..1ed653e17cb4
--- /dev/null
+++ b/lagom-cargotracker/shipping-api/src/main/java/sample/cargotracker/shipping/api/AbstractLeg.java
@@ -0,0 +1,29 @@
+package sample.cargotracker.shipping.api;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.lightbend.lagom.javadsl.immutable.ImmutableStyle;
+
+import java.util.Date;
+
+import org.immutables.value.Value;
+
+@Value.Immutable
+@ImmutableStyle
+@JsonDeserialize(as = Leg.class)
+public interface AbstractLeg {
+
+ @Value.Parameter
+ String getId();
+
+ @Value.Parameter
+ String getCargoId();
+
+ @Value.Parameter
+ String getLocation();
+
+ @Value.Parameter
+ Date getArrivalTime();
+
+ @Value.Parameter
+ Date getDepartureTime();
+}
diff --git a/lagom-cargotracker/shipping-api/src/main/java/sample/cargotracker/shipping/api/ShippingService.java b/lagom-cargotracker/shipping-api/src/main/java/sample/cargotracker/shipping/api/ShippingService.java
new file mode 100644
index 000000000000..73cabf2b5a5d
--- /dev/null
+++ b/lagom-cargotracker/shipping-api/src/main/java/sample/cargotracker/shipping/api/ShippingService.java
@@ -0,0 +1,73 @@
+package sample.cargotracker.shipping.api;
+
+import static com.lightbend.lagom.javadsl.api.Service.named;
+import static com.lightbend.lagom.javadsl.api.Service.restCall;
+
+import akka.Done;
+import akka.NotUsed;
+import com.lightbend.lagom.javadsl.api.Descriptor;
+import com.lightbend.lagom.javadsl.api.Service;
+import com.lightbend.lagom.javadsl.api.ServiceCall;
+import com.lightbend.lagom.javadsl.api.transport.Method;
+
+/**
+ * The shipping service interface.
+ *
+ * This describes everything that Lagom needs to know about how to serve and consume the ShippingService.
+ */
+public interface ShippingService extends Service {
+
+ /**
+ * Example: curl -H "Content-Type: application/json" -X POST -d
+ * '{
+ * "itinerary": {
+ * "id": "1",
+ * "cargoId": "1",
+ * "origin": "Gothom City",
+ * "destination": "Metropolis",
+ * }
+ * }' http://localhost:9000/api/itinerary
+ */
+ ServiceCall createItinerary();
+
+ /**
+ * Adds a leg to an existing itinerary.
+ *
+ * The String here is the itinerary id.
+ *
+ * {@code curl -H "Content-Type: application/json" -X POST -d
+ * '{
+ * "leg": {
+ * "id": "1",
+ * "cargoId": "cargoId",
+ * "location": "derp",
+ * "arrivalTime": arrivalTime,
+ * "departureTime": departureTime
+ * }
+ * }' http://localhost:9000/api/itinerary/1/leg
+ * }
+ */
+ ServiceCall addLeg();
+
+ @Override
+ default Descriptor descriptor() {
+ // @formatter:off
+ return named("shippingService").with(
+ restCall(Method.POST, "/api/itinerary", createItinerary()).withAutoAcl(true),
+ restCall(Method.POST, "/api/itinerary/:itinerary/leg", addLeg()).withAutoAcl(true)
+ );
+ // @formatter:on
+
+ /*
+ {
+ "leg": {
+ "id": "1",
+ "cargoId": "cargoId",
+ "location": "derp",
+ "arrivalTime": "2002-05-30T09:30:10Z",
+ "departureTime": "2002-05-30T010:30:10Z"
+ }
+ }
+ */
+ }
+}
diff --git a/lagom-cargotracker/shipping-api/src/main/resources/application.conf b/lagom-cargotracker/shipping-api/src/main/resources/application.conf
new file mode 100644
index 000000000000..e0f7afc396bc
--- /dev/null
+++ b/lagom-cargotracker/shipping-api/src/main/resources/application.conf
@@ -0,0 +1 @@
+play.http.errorHandler=sample.cargotracker.shipping.ErrorHandler
diff --git a/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/AbstractItineraryState.java b/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/AbstractItineraryState.java
new file mode 100644
index 000000000000..db06a44dee6b
--- /dev/null
+++ b/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/AbstractItineraryState.java
@@ -0,0 +1,56 @@
+package sample.cargotracker.shipping.impl;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.lightbend.lagom.javadsl.immutable.ImmutableStyle;
+import com.lightbend.lagom.serialization.CompressedJsonable;
+import java.util.List;
+import java.time.LocalDateTime;
+import org.immutables.value.Value;
+import org.pcollections.PCollection;
+import org.pcollections.PSequence;
+import sample.cargotracker.shipping.api.Leg;
+
+/**
+ * The state for the itinerary entity.
+ */
+@Value.Immutable
+@ImmutableStyle
+@JsonDeserialize(as = ItineraryState.class)
+public abstract class AbstractItineraryState implements CompressedJsonable {
+
+ /**
+ * The itinerary id.
+ */
+ @Value.Parameter
+ public abstract String getId();
+
+ /**
+ * The cargo id.
+ */
+ @Value.Parameter
+ public abstract String getCargoId();
+
+ /**
+ * The cargo origin.
+ */
+ @Value.Parameter
+ public abstract String getOrigin();
+
+ /**
+ * The cargo destination.
+ */
+ @Value.Parameter
+ public abstract String getDestination();
+
+ /**
+ * The legs for the itinerary.
+ */
+ @Value.Parameter
+ public abstract PSequence getLegs();
+
+ /**
+ * When the greeting last changed.
+ */
+ @Value.Parameter
+ public abstract LocalDateTime getTimestamp();
+}
diff --git a/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ItineraryEntity.java b/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ItineraryEntity.java
new file mode 100644
index 000000000000..8e372bfe393e
--- /dev/null
+++ b/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ItineraryEntity.java
@@ -0,0 +1,107 @@
+package sample.cargotracker.shipping.impl;
+
+import com.lightbend.lagom.javadsl.persistence.PersistentEntity;
+import java.time.LocalDateTime;
+import java.util.Optional;
+
+import akka.Done;
+import org.pcollections.PSequence;
+import org.pcollections.TreePVector;
+import sample.cargotracker.shipping.api.Leg;
+
+public class ItineraryEntity extends PersistentEntity {
+ /**
+ * An entity can define different behaviours for different states, but it will always start with an
+ * initial behaviour. This entity only has one behaviour.
+ */
+ @Override
+ public Behavior initialBehavior(Optional snapshotState) {
+
+ /*
+ * Behaviour is defined using a behaviour builder. The behaviour builder starts with a state, if this
+ * entity supports snapshotting (an optimisation that allows the state itself to be persisted to combine
+ * many events into one), then the passed in snapshotState may have a value that can be used.
+ *
+ * Otherwise, the default state is to use a dummy Cargo with and id of 0L..
+ */
+ BehaviorBuilder b = newBehaviorBuilder(snapshotState.orElse(
+ ItineraryState.builder()
+ .id("")
+ .cargoId("")
+ .origin("")
+ .destination("")
+ .legs(TreePVector.empty())
+ .timestamp(LocalDateTime.now()).build()));
+
+ /*
+ * Command handler for the create itinerary command.
+ */
+ b.setCommandHandler(CreateItinerary.class,
+ (cmd, ctx) ->
+ // In response to this command, we want to first persist it as a itinerary created event
+ ctx.thenPersist(
+ ItineraryCreated.of(
+ cmd.getItinerary().getId(),
+ cmd.getItinerary().getCargoId(),
+ cmd.getItinerary().getOrigin(),
+ cmd.getItinerary().getDestination()),
+ // Then once the event is successfully persisted, we respond with done.
+ evt -> ctx.reply(Done.getInstance())
+ )
+ );
+
+ /*
+ * Command handler for the create itinerary command.
+ */
+ b.setCommandHandler(AddLeg.class,
+ (cmd, ctx) ->
+ // In response to this command, we want to first persist it as a itinerary created event
+ ctx.thenPersist(
+ LegAdded.of(
+ cmd.getLeg().getId(),
+ cmd.getLeg().getCargoId(),
+ cmd.getLeg().getLocation(),
+ cmd.getLeg().getArrivalTime(),
+ cmd.getLeg().getDepartureTime()),
+ // Then once the event is successfully persisted, we respond with done.
+ evt -> ctx.reply(Done.getInstance())
+ )
+ );
+
+ /*
+ * Event handler for the itinerary created event.
+ */
+ b.setEventHandler(LegAdded.class,
+ evt -> {
+ final Leg leg = Leg.builder()
+ .id(evt.getId())
+ .cargoId(evt.getCargoId())
+ .location(evt.getLocation())
+ .arrivalTime(evt.getArrivalTime())
+ .departureTime(evt.getDepartureTime())
+ .build();
+ return state().withLegs(state().getLegs().plus(leg));
+ }
+ );
+
+ /*
+ * Event handler for the itinerary created event.
+ */
+ b.setEventHandler(ItineraryCreated.class,
+ // We simply update the current state to use the greeting message from the event.
+ evt -> state()
+ .withId(evt.getId())
+ .withCargoId(evt.getCargoId())
+ .withOrigin(evt.getOrigin())
+ .withDestination(evt.getDestination())
+ .withTimestamp(LocalDateTime.now())
+ );
+
+ /*
+ * We've defined all our behaviour, so build and return it.
+ */
+ return b.build();
+ }
+
+
+}
diff --git a/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ShippingCommand.java b/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ShippingCommand.java
new file mode 100644
index 000000000000..579f21d35e48
--- /dev/null
+++ b/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ShippingCommand.java
@@ -0,0 +1,49 @@
+package sample.cargotracker.shipping.impl;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.lightbend.lagom.javadsl.immutable.ImmutableStyle;
+import com.lightbend.lagom.javadsl.persistence.PersistentEntity;
+import com.lightbend.lagom.serialization.CompressedJsonable;
+import com.lightbend.lagom.serialization.Jsonable;
+import java.util.Optional;
+import org.immutables.value.Value;
+
+import sample.cargotracker.shipping.api.Itinerary;
+
+import akka.Done;
+import sample.cargotracker.shipping.api.Leg;
+
+/**
+ * This interface defines all the commands that the Itinerary entity supports.
+ *
+ * By convention, the commands should be inner classes of the interface, which makes it simple to get a
+ * complete picture of what commands an entity supports.
+ */
+public interface ShippingCommand extends Jsonable {
+
+ /**
+ * A command to create an itinerary.
+ *
+ * It has a reply type of {@link akka.Done}, which is sent back to the caller when all the events
+ * emitted by this command are successfully persisted.
+ */
+ @Value.Immutable
+ @ImmutableStyle
+ @JsonDeserialize(as = CreateItinerary.class)
+ public interface AbstractCreateItinerary extends ShippingCommand, CompressedJsonable,
+ PersistentEntity.ReplyType {
+
+ @Value.Parameter
+ Itinerary getItinerary();
+ }
+
+ @Value.Immutable
+ @ImmutableStyle
+ @JsonDeserialize(as = AddLeg.class)
+ public interface AbstractAddLeg extends ShippingCommand, CompressedJsonable,
+ PersistentEntity.ReplyType {
+
+ @Value.Parameter
+ Leg getLeg();
+ }
+}
diff --git a/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ShippingEvent.java b/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ShippingEvent.java
new file mode 100644
index 000000000000..75e12b3aa29b
--- /dev/null
+++ b/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ShippingEvent.java
@@ -0,0 +1,63 @@
+package sample.cargotracker.shipping.impl;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.lightbend.lagom.javadsl.immutable.ImmutableStyle;
+import com.lightbend.lagom.serialization.Jsonable;
+import org.immutables.value.Value;
+import org.immutables.value.Value.Immutable;
+
+import java.util.Date;
+
+/**
+ * This interface defines all the events that the Shipping entity supports.
+ *
+ * By convention, the events should be inner classes of the interface, which makes it simple to get a
+ * complete picture of what events an entity has.
+ */
+public interface ShippingEvent extends Jsonable {
+
+ /**
+ * An event that represents a new shipping itinerary.
+ */
+ @Immutable
+ @ImmutableStyle
+ @JsonDeserialize(as = ItineraryCreated.class)
+ interface AbstractItineraryCreated extends ShippingEvent {
+
+ @Value.Parameter
+ String getId();
+
+ @Value.Parameter
+ String getCargoId();
+
+ @Value.Parameter
+ String getOrigin();
+
+ @Value.Parameter
+ String getDestination();
+ }
+
+ /**
+ * An event that represents a leg being added.
+ */
+ @Immutable
+ @ImmutableStyle
+ @JsonDeserialize(as = LegAdded.class)
+ interface AbstractLegAdded extends ShippingEvent {
+
+ @Value.Parameter
+ String getId();
+
+ @Value.Parameter
+ String getCargoId();
+
+ @Value.Parameter
+ String getLocation();
+
+ @Value.Parameter
+ Date getArrivalTime();
+
+ @Value.Parameter
+ Date getDepartureTime();
+ }
+}
diff --git a/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ShippingServiceImpl.java b/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ShippingServiceImpl.java
new file mode 100644
index 000000000000..4bf53ba3433f
--- /dev/null
+++ b/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ShippingServiceImpl.java
@@ -0,0 +1,53 @@
+package sample.cargotracker.shipping.impl;
+
+import akka.Done;
+import akka.NotUsed;
+import com.lightbend.lagom.javadsl.api.ServiceCall;
+import com.lightbend.lagom.javadsl.persistence.PersistentEntityRef;
+import com.lightbend.lagom.javadsl.persistence.PersistentEntityRegistry;
+import javax.inject.Inject;
+
+import sample.cargotracker.registration.api.Cargo;
+import sample.cargotracker.registration.api.RegistrationService;
+import sample.cargotracker.shipping.api.Itinerary;
+import sample.cargotracker.shipping.api.Leg;
+import sample.cargotracker.shipping.api.ShippingService;
+
+import java.util.concurrent.CompletionStage;
+
+/**
+ * Implementation of the ShippingService.
+ */
+public class ShippingServiceImpl implements ShippingService {
+
+ private final PersistentEntityRegistry persistentEntityRegistry;
+ private final RegistrationService registrationService;
+
+ @Inject
+ public ShippingServiceImpl(PersistentEntityRegistry persistentEntityRegistry, RegistrationService registrationService) {
+ this.persistentEntityRegistry = persistentEntityRegistry;
+ this.registrationService = registrationService;
+ persistentEntityRegistry.register(ItineraryEntity.class);
+ }
+
+ @Override
+ public ServiceCall addLeg() {
+ return (id, request) -> {
+ CompletionStage response = registrationService.getRegistration().invoke(request.getCargoId(), NotUsed.getInstance());
+ PersistentEntityRef itinerary = persistentEntityRegistry.refFor(ItineraryEntity.class, id);
+ return itinerary.ask(AddLeg.of(request));
+ };
+ }
+
+ @Override
+ public ServiceCall createItinerary() {
+ return (id, request) -> {
+ // Look up the itinerary for the given ID.
+ PersistentEntityRef itinerary =
+ persistentEntityRegistry.refFor(ItineraryEntity.class, request.getId());
+ // Tell the entity to use the greeting message specified.
+ return itinerary.ask(CreateItinerary.of(request));
+ };
+ }
+
+}
diff --git a/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ShippingServiceModule.java b/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ShippingServiceModule.java
new file mode 100644
index 000000000000..0de7ae842472
--- /dev/null
+++ b/lagom-cargotracker/shipping-impl/src/main/java/sample/cargotracker/shipping/impl/ShippingServiceModule.java
@@ -0,0 +1,17 @@
+package sample.cargotracker.shipping.impl;
+
+import com.google.inject.AbstractModule;
+import com.lightbend.lagom.javadsl.server.ServiceGuiceSupport;
+import sample.cargotracker.registration.api.RegistrationService;
+import sample.cargotracker.shipping.api.ShippingService;
+
+/**
+ * The module that binds the ShippingService so that it can be served.
+ */
+public class ShippingServiceModule extends AbstractModule implements ServiceGuiceSupport {
+ @Override
+ protected void configure() {
+ bindServices(serviceBinding(ShippingService.class, ShippingServiceImpl.class));
+ bindClient(RegistrationService.class);
+ }
+}
diff --git a/lagom-cargotracker/shipping-impl/src/main/resources/application.conf b/lagom-cargotracker/shipping-impl/src/main/resources/application.conf
new file mode 100644
index 000000000000..c6154749372c
--- /dev/null
+++ b/lagom-cargotracker/shipping-impl/src/main/resources/application.conf
@@ -0,0 +1 @@
+play.modules.enabled += sample.cargotracker.shipping.impl.ShippingServiceModule