Martin Krasser's Blog

Building an Event-Sourced Web Application - Part 2: Projections, Persistence, Consumers and Web Interface

2012-01-20T17:34:00.013+01:00

NOTE: This is a draft. In its current state, this blog post gives an overview of the latest event-sourcing example application developments. It also serves as a guide to the relevant code sections with more detailed descriptions to follow. One topic that is already covered in more detail is Service Layer Enhancements.

Overview

A few weeks ago I started a blog post series to summarize my experiences in building an event-sourced web application using Scala and Akka. This was done based on an example application (see branch part-1). There, I gave an overview of the application architecture and presented some details of the immutable domain model and the service layer. Since then, the example application has been extended (see branch part-2) with a number of new features and enhancements. Here's an overview:

Enhancements are:

The STM-based state management was completely revised and generalized into the traits UpdateProjection and EventProjection. An UpdateProjection is similar to an Akka Agent: it applies state transition functions asynchronously and can participate in STM transactions. The major difference is that an UpdateProjection is specialized on domain object updates and can log captured events to a persistent event log. By default, update events are logged before state changes are visible via STM references (this is an important difference to the service layer of part 1). UpdateProjection implementors (such as the example application's InvoiceService) are domain event producers. EventProjection implementors, on the other hand, are domain event consumers. They internally use plain Akka Agents to manage state and derive new state values from received domain events (using an application-defined event projection function). EventProjection implementors are usually components that manage read models or coordinate business processes in event-driven architectures, for example.

The domain model was enhanced by introducing the domain classes DraftInvoice, SentInvoice and PaidInvoice to represent the states an invoice can have. Valid state transitions are defined by the methods on these domain classes and can therefore be checked by the compiler. This approach is explained more detailed here although the implementation used in our example application slightly differs. In part 1, we only had a single Invoice class and valid state transitions had to be checked at runtime.

New features include:

A revised EventLog trait together with two implementations: JournalioEventLog is based on Journal.IO and BookkeeperEventLog on Apache BookKeeper. An EventLog supports synchronous and asynchronous appending of events as well as iterating over stored events, either from the beginning or from an application-defined position in the event history. Event log entries are also assigned sequence numbers so that event consumers can detect gaps in event streams or re-order (resequence) them, if needed.

Event consumers. One example is InvoiceStatistics, an EventProjection that derives invoice update statistics from domain events. Here, a separate read model is used (following the CQRS pattern) to serve invoice statistic queries. Another example is InvoiceReplicator, an EventProjection that simply reconstructs the invoice map (as maintained by the InvoiceService) from invoice events at a different location. It can be used to replicate application state across different nodes and to serve (eventually consistent) reads. The replicated state could also be used by a snapshot service to take snapshots of application state. InvoiceReplicator needs to receive events in the correct order and is therefore configured to resequence the received event stream. A third example is the PaymentProcess. It coordinates the activities of InvoiceService and PaymentService. Instead of having these services sending commands to each other, it is the PaymentProcess that sends commands to (i.e. calls methods on) these services in reaction to domain events. This event-driven approach to implementing business processes not only decouples the services from each other but also lets other components extend (or monitor) the business process by subscribing to and reacting on the relevant domain events. The PaymentProcess is currently stateless. Processes that need to maintain state should implement EventProjection and recover the process state during application start (or failover) from the event history.

A RESTful web interface for invoices and invoice statistics with support for HTML, XML and JSON representations. These can be negotiated with the HTTP Accept header. The web layer is based on the Jersey web framework (the JAX-RS reference implementation). HTML representations are rendered with the Scalate template engine. The mapping between XML and JSON representations and immutable domain classes is based on JAXB annotations. A JAXB-based XML provider must be supported by any JAX-RS implementation but Jersey additionally comes with a JAXB-based JSON provider so that the same metadata (JAXB annotations) can be used to generate both XML and JSON representations. Following some simple rules, it is possible to JAXB-annotate Scala case classes without polluting them with Java collection types or getters and setters. One major drawback of the current JAX-RS specification is that it doesn't support asynchronous responses yet. This will change with JAX-RS 2.0 and then we can make full use of the asynchronous InvoiceService responses.

A communication Channel for connecting domain event producers to consumers. The example application provides a SimpleChannel implementation for local communication. Alternative implementations, for example, could connect to a distributed event bus to communicate events across components of a distributed application.

Running the example application

The example application can be started with

sbt run-main dev.example.eventsourcing.server.Webserver

Two classes relevant for starting the application are:

Appserver: configures the event log, services, read models and processes and connects them via an event channel. It also recovers application state from the event history.

Webserver: configures the web service layer and starts an embedded web server (Jetty).

To experiment with the BookKeeper based event log, you need to replace JournalioEventLog with BookkeeperEventLog in Appserver and additionally start a test BookKeeper instance with

sbt run-main dev.example.eventsourcing.server.Zookeeper
sbt run-main dev.example.eventsourcing.server.Bookkeeper

Examples how to interact with the RESTful web interface can be found here.

Service Layer Enhancements

In the service layer implementation from part 1 we've seen how to keep the order of logged events in correspondence with the order of updates to the application state. We used a transient event log that could participate in STM transactions. After the transaction commits, the events from the transient event log have been transferred to a persistent event log. A drawback of this approach is that one can loose updates in case of crashes after an STM reference has been updated but before the changes have been written to the persistent event log. This can lead to situations where clients can already see application state that cannot be recovered from the event log. While some applications may tolerate this, others may require that any visible application state must be fully recoverable from the event log. Therefore, an alternative approach must be chosen.

We need a way to write events, captured during domain object updates, to a persistent event log before the STM reference is updated. But writing to the persistent event log must be done outside an STM transaction for reasons explained in part 1. Updates must also be based on the current (i.e. latest) application state. We therefore need to

Get the current state value from a transactional reference (STM transaction 1)

Update domain object(s) obtained from the current state (no STM transaction)

Write the captured update event(s) to a persistent event log (no STM transaction)

Compute a new state value from the domain object update and write it to the transactional reference (STM transaction 2)

Since steps 1-4 are not a single transaction we must prevent their concurrent execution. This can be achieved using a single actor, for example. This actor is the single writer to the transactional reference. This is actually very similar to how Akka Agents work internally. The main difference in our case is that the computation of a new state value occurs outside an STM transaction, whereas Akka Agents apply update functions within STM transactions. This difference allows us to have side effects, such as writing to a persistent event log. The example application provides the above functionality in the UpdateProjection trait:

Instances of UpdateProjection manage (part of) application state with a transactional ref of type Ref[S] where S is the state value type. Clients concurrently read application state via currentState. Sequential writes to the transactional ref are done exclusively by the updater actor (more on write concurrency below).

UpdateProjection implementors change application state with the transacted method. The update parameter is a function that computes a domain object Update[Event, B] from current state S where B is a domain object type. The update result (either Success[B] or Failure[DomainError]) is returned as future value from the transacted method.

The update function and the underlying Future implementation object (promise) are sent to the updater actor with an ApplyUpdate message. The updater then reads the current state and applies the update function to it. If the update succeeds, it writes the captured events to an EventLog and projects the update result onto the current state. The projection is done with the project function. It creates a new state value from the current state and the update result. The new state value is then finally set on the transactional ref and the promise is completed with the update result.

Furthermore, the transacted method can participate in STM transactions. If there's an enclosing transaction, the updater actor will only be triggered if the enclosing transaction successfully commits. If there's no enclosing transaction the updater actor will always be triggered.

The example application uses UpdateProjection to implement the stateful InvoiceService. The state is of type Map[String, Invoice] i.e. a single map containing draft, sent and paid invoices. Here's a simplified version of InvoiceService:

The updateInvoice method uses the transacted method of the UpdateProjection trait. It tries to get an invoice with given invoiceId from the current state and applies the supplied update function f to it. The updateInvoice method is used by updateDraftInvoice for updating draft invoices in the invoice map. The updateDraftInvoice method is used by the service methods addInvoiceItem and sendInvoiceTo. Adding an item to an existing draft invoice yields a future value of an updated draft invoice (return type Future[DomainValidation[DraftInvoice]]). Sending an existing draft invoice, on the other hand, causes a state transition of that invoice to a sent invoice (return type Future[DomainValidation[SentInvoice]]) i.e. the service methods make use of the newly introduced domain object types. The InvoiceService must also implement the abstract members project, initialState and eventLog (declared by Projection and UpdateProjection).

The project implementation projects an updated invoice onto the current state by simply adding it to the map of invoices (replacing an old invoice if present).

An initialState (empty map by default) and an EventLog instance are provided during construction of an InvoiceService.

The above InvoiceService implementation supports concurrent reads but only sequential writes to the whole invoice map. This may be sufficient for many applications but if a higher degree of write concurrency is needed, one could choose to have a separate UpdateProjection instance per invoice object (which is comparable to have a separate Akka Agent for each invoice object). This allows both concurrent reads and writes to the invoices of an application. The following snippet shows the general idea (not part of the example project).

Here, an InvoiceService maintains a map of PersistentInvoice instances where a PersistentInvoice is an UpdateProjection that contains a reference to a single (draft, sent or paid) invoice. Consequently, updates to different invoices can now be made concurrently. The projection function degenerates to a function that simply returns the updated invoice.

Event Logs

TODO

CQRS and Consistency

TODO

Business Processes

TODO

Web Interface

TODO

Building an Event-Sourced Web Application - Part 1: Domain Model, Events and State

2011-11-29T13:06:00.032+01:00

Over the last few months I was working for a project in which we built an event-sourced healthcare web application from scratch with Scala and Akka. This is the first of a series of blog posts where I want to share some of my experiences and design decisions.

The application architecture follows an architectural pattern that has been described as memory image or system prevalence: application state is kept in memory rather than written to a database. Only state changes are written to a persistent store in terms of domain events. Snapshots of application state are saved to disk at regular intervals. Application state can be reconstructed by replaying the logged events (either beginning from scratch or from a saved snapshot). We've chosen this architecture because

The state of our application easily fits within 2 GB of memory. Should this become a limitation in the future, we can easily partition the application state across several nodes (which can also be used to scale writes, if needed).

Very low latency can be achieved for read and write access to application state. To ensure atomicity, consistency and isolation, we use a Software Transactional Memory (STM). Durability is achieved with a persistent event log.

The application must be able to answer queries not only based on its current state but also based on the history of state changes. Requirements for such queries are often not known from the beginning and can therefore not be considered during initial database design. Using a persistent event log instead, one can build suitable read models any time later by replaying the whole event history.

Multiple copies of the application state can be created on other nodes by consuming domain events that have been published by a writer (or leader) node. Should the current writer go down, one can easily switch to another node to achieve high-availability.

From the memory image approach, we excluded all data that are written only once and are not modified later, such as large medical image files or clinical documents. They are stored directly on a (distributed) file system and only references to it are kept in memory.

Application Overview

The following list briefly summarizes some of the concepts and technologies used to implement the different parts of the application. In this blog post I'll focus on the domain model and the service layer. The other parts will be described in follow-up posts.

The domain model is an immutable domain model. One advantage of immutable domain objects is that you can safely share with other application components, for example, by sending them along with event messages. Immutable domain objects are also needed when using them together with Software Transactional Memory (see also this article).

The service layer provides transactional access to the application state. Application state is managed by transactional references. State values are (immutable) domain objects. State changes occur by updating the transactional references with new domain object values. For transaction management, Akka's Software Transactional Memory (STM) is used. The application's approach to state, identity and concurrency mainly follows the concepts described in State and Identity (which is part of the Clojure documentation).

The persistence layer comprises a persistent event log and snapshots of the application state. In the production system, we use Apache BookKeeper as distributed, highly-available and scalable event log. Snapshots are stored on a distributed file system.

The web layer provides a RESTful service interface to application resources (domain objects) with support for HTML, XML and JSON as representation formats. We use Jersey as web framework for implementing the RESTful service interface together with Scalate to render the HTML representations of domain objects. For XML and JSON bindings we use JAXB annotations on Scala case classes - seriously :)

Read models are used to serve more complex queries that cannot be (efficiently) answered by using the current application state. Read models are also event-sourced. Their structure is optimized to answer complex queries in a very efficient way. The approach to separate write models from read models is called Command Query Responsibility Segregation (CQRS). Read models can be stored in-memory (and reconstructed by replaying events) or persisted to a database. We use Akka agents to maintain in-memory read models.

Business process executors are stateful event listeners that implement long-running business processes. In reaction to domain events, they may change application state (via the service layer), coordinate changes across aggregate roots or interact with external services. We use Akka actors to implement business process executors.

Versioning of domain objects is used to support conditional updates. Only if the current version of a domain object matches a client-supplied version number, an update can proceed. We added this form of optimistic concurrency control in addition to that of STM transactions.

For distributed coordination, we use Apache Zookeeper. There's one node in our distributed system that performs state changes (the current writer or leader). Should the current leader go down, another leader is elected using a leader election algorithm implemented on top of Zookeeper. All nodes in the distributed system can serve (eventually consistent) reads. Strongly consistent reads can only be served by the current leader. We can therefore easily scale eventually consistent reads which is the majority of reads in our application.

Example application

There's also an example project on github that demonstrates how to combine the contents of the following sections to a running application. I'll extend the application as I make progress on this blog post series. The branch referred to by this blog post is part-1.

The example application will cover some elements of our production system but not all of them. To keep things simple, I omitted performance optimizations and decided to use an over-simplified domain model from an other than the healthcare domain.

Domain Model

The approach to implement an immutable domain model was taken from the excellent article series Towards an Immutable Domain Model by Erik Rozendaal. In the following I'll briefly summarize this approach (with some modifications). For a more detailed description, please read through these articles series.

The domain model of the example application is defined by the case classes Invoice, InvoiceItem and InvoiceAddress (see Invoice.scala). Invoice is the aggregate root with methods to add an InvoiceItem, set a discount and send an Invoice (to a destination defined by InvoiceAddress).

The methods addItem, setDiscount and sendTo generate the domain events InvoiceItemAdded, InvoiceDiscountSet and InvoiceSent, respectively (see later). These are then handled by the handle method. The handle method creates Invoice copies with updated members.

In addition to generating domain events, the methods addItem, setDiscount and sendTo not only execute business logic (by checking preconditions, for example) but also capture the generated events. Generated events are captured with the Update monad (a state monad) which is the return type of these methods. This is shown for sendTo.

A successful update adds the generated InvoiceSent event and the updated Invoice to the returned Update instance (using the update method from the EventSourced trait). A failed update is reported with a DomainError (using Update.reject). Update results are instances of Validation (either Success or Failure) which can be obtained by calling result() on the monad

where Success contains the updated Invoice and Failure contains the reported error. To get access to the captured events, call the result method with a callback function.

This function is only called for successful updates with the captured events and the updated Invoice. We will see a concrete usage example later. Since Update is a monad, we can also chain domain object updates with a for-comprehension.

Only if all individual updates are successful, the overall update will be successful. If one or more individual updates fail, the overall update will fail. Finally, we want to be able to reconstruct an Invoice object from the history of events. This can be done with the Invoice.handle(events: List[InvoiceEvent]) method on the Invoice companion object, as shown in the following example.

Having these domain model properties in place, we can now use immutable Invoice objects to define application state and use captured events to write them to a (persistent) event log.

Service Layer

Application state is managed in the application's service layer. Here, InvoiceService maintains a map of Invoice objects where map keys are invoice ids.

To control concurrent access to the invoices map, a transactional reference (akka.stm.Ref) is used. It is part of Akka's STM (Multiverse) but rewriting the example using Scala's STM shouldn't be a big deal. Using a single transactional reference for all Invoice objects is a rather naive approach because it can easily lead to high contention on the invoicesRef, but it keeps the example as simple as possible. (Using a single transactional reference for all Invoice objects is a reasonable approach only if updates to different Invoice objects depend on each other which is usually not the case, or if there aren't many concurrent writes. We will see better approaches, causing less contention, later in this section). An alternative approach for managing state is using actors but I'll leave that for another blog post.

Changes to the application state are made inside an atomic {...} block, as shown for the addInvoiceItem method (see also InvoiceService.scala).

If any two updates to the Invoice map are conflicting, one update will succeed, the other one will be retried by re-executing the atomic block. Since we are using immutable domain objects, retrying updates is not an issue. What's still missing is to log the captured events (obtained from the Update monad) to a persistent event log. In the following, we require that the order of logged events must correspond to the order of updates to the application state (otherwise we could get problems during a replay of events). We also make the assumption that our persistent event log cannot directly participate in STM transactions (which is the case for Apache BookKeeper, for example). We could try to:

Write captured events to the persistent event log within a transaction. Writing to a persistent event log involves IO operations (side-effects) that cannot be rolled back. Should the STM retry the transaction, the atomic block is re-executed and the captured events would be written to the event log again and this is not what we want.

Write the captured events after the STM transaction commits with the thread that started the transaction. This would solve the problem of redundantly logged events but then we cannot guarantee any more that the order of logged events still corresponds to the order of updates to the application state. This is because the writes could be done by different threads which introduces a race.

So these two approaches don't work. One possible solution is the following approach:

Write the captured events to a transient, transactional event log inside the atomic block. This ensures that events are not redundantly logged and the order of events corresponds to the order of updates to the application state. The simplest possible transient, transactional event log is a Ref[List[Event]].

Transfer the logged events from the transactional event log to a persistent event log by preserving the order of events. Preserving the order of events can be achieved with a single thread (or actor) that reads from the transactional event log and writes events to the persistent event log. Any time an STM transaction commits we need to trigger this thread (or actor) to do the transfer.

Let's say this functionality is available through an EventLog trait:

where

the log method adds events to the transactional eventsRef (must be called within a transaction) and

the store* methods transfer the events from the eventsRef to the persistent event log (must be called after the transaction commits). The store method waits for the transfer to complete whereas storeAsync returns immediately.

Having an implementation of EventLog in place, it can be used as follows:

Within the transaction, captured events are added to the transactional event log. After the transaction successfully commits, the events are transferred to the persistent event log. This is done within the deferred block which is executed only once after commit. Here, we don't wait for the events being persisted (storeAsync). We could also extend the storeAsync and addInvoiceItem method signatures to let clients provide an asynchronous completion callback function in order to be notified when events have been written successfully (or an error occurred during writing). A production-ready implementation of EventLog should also provide strategies to recover from errors writing to the persistent event log. In a follow-up blog post I'll show an implementation of EventLog that uses Apache BookKeeper (update: I'll also show how to do write-ahead logging which first writes events to the event log and then, if writing was successful, updates the invoices map. This can be done by queueing up updates). In its current state the example application has an EventLog implementation that stores events in memory, for testing purposes (see TestEventLog.scala)

The InvoiceService also provides methods for reading invoices. Supporting consistent reads is as easy as

Here invoicesRef() implicitly starts a new transaction, so we don't need to make the read operation within an atomic {...} block. There are some situations where consistent reads are needed by clients (for example, in a web application during a post/redirect/get cycle where invoices must be immediately available for reading after their creation). In this case, the InvoiceService should be used. In other situations, eventually consistent reads are sufficient. In that case, we wouldn't use the InvoiceService to obtain Invoice objects, we'd rather obtain it from a separate (event-sourced) read model that is asynchronously generated from published events (using CQRS). This will be shown in a follow-up blog post.

Finally, lets look at some options how contention on the invoicesRef (and the transactional event log) can be reduced (in situations with many concurrent writes). We can say that

Updates to different Invoice entities are independent, so we don't need to care about ordering of events in this case.

Updates to the same Invoice entity are dependent, so the order of logged events must correspond to the order of updates to that entity.

This means that we can use transactional references for each individual Invoice entity. Consequently, updates to different invoice entities do not interfere with each other.

The transactional reference for the whole map is only needed for the case that new invoices are concurrently added to the map (and to ensure consistent reads, of course). We could also use a separate event log for each Ref[Invoice], so that there's no contention on the transactional event log for independent updates.

In this case, different transactional event logs would share the same persistent event log backend.

Summary

We started to build an event-sourced application whose state is kept entirely in memory, only state changes are persisted to an event log in terms of domain events. Application state is defined by immutable domain objects which are accessed through transactional references. A state change means updating the transactional references with new domain objects (new state value) within a STM transaction. Events that have been generated during domain object updates are written to a transient, transactional event log. Transfer to a persistent event log occurs once the transaction commits. It was also shown how the order of logged events can be kept consistent with the order of state changes. This is important when application state must be recovered from the event history (for example, during application start or failover to another node).

Akka Producer Actors: New Features and Best Practices

2011-02-28T07:53:00.009+01:00

In a previous post I wrote about new features and best practices for Akka consumer actors. In this post, I'll cover Akka producer actors. For the following examples to compile and run, you'll need the current Akka 1.1-SNAPSHOT.

Again, I assume that you already have a basic familiarity with Akka, Apache Camel and the akka-camel integration module. If you are new to it, you may want to read the Akka and Camel chapter (free pdf) of the Camel in Action book or the Introduction section of the official akka-camel documentation first.

Basic usage

Akka producer actors can send messages to any Camel endpoint, provided that the corresponding Camel component is on the classpath. This allows Akka actors to interact with external systems or other components over a large number of protocols and APIs.

Let's start with a simple producer actor that sends all messages it receives to an external HTTP service and returns the response to the initial sender. For sending messages over HTTP we can use the Camel jetty component which features an asynchronous HTTP client.

Concrete producer actors inherit a default implementation of Actor.receive from the Producer trait. For simple use cases, only an endpoint URI must be defined. Producer actors also require a started CamelContextManager for working properly. A CamelContextManager is started when an application starts a CamelService e.g. via CamelServiceManager.startCamelService or when starting the CamelContextManager directly via

The latter approach is recommended when an application uses only producer actors but no consumer actors. This slightly reduces the overhead when starting actors. After starting the producer actor, clients can interact with the HTTP service via the actor API.
kra

Here, !! is used for sending the message and waiting for a response. Alternatively, one can also use ! together with an implicit sender reference.

In this case the sender will receive an asynchronous reply from the producer actor. Before, the producer actor itself receives an asynchronous reply from the jetty endpoint. The asynchronous jetty endpoint doesn't block a thread waiting for a response and the producer actor doesn't do that either. This is important from a scalability perspective, especially for longer-running request-response cycles.

By default, a producer actor initiates an in-out message exchange with its Camel endpoint i.e. it expects a response from it. If a producer actor wants to initiate an in-only message exchange then it must override the oneway method to return true. The following example shows a producer actor that initiates an in-only message exchange with a JMS endpoint.

This actor adds any message it receives to the test JMS queue. By default, producer actors that are configured with oneway = true don't reply. This behavior is defined in the Producer.receiveAfterProduce method which is implemented as follows.

The receiveAfterProduce method has the same signature as Actor.receive and is called with the result of the message exchange with the endpoint (please note that in-only message exchanges with Camel endpoints have a result as well). The result type for successful message exchanges is Message, for failed message exchanges it is Failure (see below).

Concrete producer actors can override this method. For example, the following producer actor overrides onReceiveAfterProduce to reply with a constant "done" message.

The result of the message exchange with the JMS endpoint is ignored (case _).

Failures

Messages exchanges with a Camel endpoint can fail. In this case, onReceiveAfterProduce is called with a Failure message containing the cause of the failure (a Throwable). Let's extend the HttpProducer usage example to deal with failure responses.

In addition to a failure cause, a Failure message can also contain endpoint-specific headers with failure details such as the HTTP response code, for example. When using ! instead of !!, together with an implicit sender reference (as shown in the previous section), that sender will then receive the Failure message asynchronously. The JmsReplyingProducer example can also be extended to return more meaningful responses: a "done" message only on success and an error message on failure.

Failed message exchanges never cause the producer actor to throw an exception during execution of receive. Should Producer implementations want to throw an exception on failure (for whatever reason) they can do so in onReceiveAfterProduce.

In this case failure handling should be done in combination with a supervisor (see below).

Let's look at another example. What if we want

to throw an exception on failure (instead of returning a Failure message) but to respond with a normal Message on success? In this case, we need to use self.senderFuture inside onReceiveAfterProduce and complete it with an exception.

Forwarding results

Another option to deal with message exchange results inside onReceiveAfterProduce is to forward them to another actor. Forwarding a message also forwards the initial sender reference. This allows the receiving actor to reply to the initial sender.

With producer actors that forward message exchange results to other actors (incl. other producer actors) one can build actor-based message processing pipelines that integrate external systems. In combination with consumer actors, this could be extended towards a scalable and distributed enterprise service bus (ESB) based on Akka actors ... but this is a topic for another blog post.

Correlation identifiers

The Producer trait also supports correlation identifiers. This allows clients to correlate request messages with asynchronous response messages. A correlation identifier is a message header that can be set by clients. The following example uses the correlation identifier (or message exchange identifier) 123.

An asynchronous response (Message or Failure) from httpProducer will contain that correlation identifier as well.

Fault-tolerance

A failed message exchange by default does not cause a producer actor to throw an exception. However, concrete producer actors may decide to throw an exception inside onReceiveAfterProduce, for example, or there can be a system-level Camel problem that causes a runtime exception. An application that wants to handle these exceptions should supervise its producer actors.

The following example shows how to implement a producer actor that replies to the initial sender with a Failure message when it is restarted or stopped by a supervisor.

To handle restart callbacks, producer actors must override the preRestartProducer method instead of preRestart. The preRestart method is implemented by the Producer trait and does additional resource de-allocation work after calling preRestartProducer. More information about replies within preRestart and postStop can be found in my previous blog post about consumer actors.

Akka Consumer Actors: New Features and Best Practices

2011-02-17T13:58:00.028+01:00

In this blog post I want to give some guidance how to implement consumer actors with the akka-camel module. Besides basic usage scenarios, I will also explain how to make consumer actors fault-tolerant, redeliver messages on failure, deal with bounded mailboxes etc. The code examples shown below require the current Akka 1.1-SNAPSHOT to compile and run.

In the following, I assume that you already have a basic familiarity with Akka, Apache Camel and the akka-camel integration module. If you are new to it, you may want to read the Akka and Camel chapter (free pdf) of the Camel in Action book or the Introduction section of the official akka-camel documentation first.

Basic usage

Akka consumer actors can receive messages from any Camel endpoint, provided that the corresponding Camel component is on the classpath. This allows clients to interact with Akka actors over a large number of protocols and APIs.

Camel endpoints either initiate in-only (one-way) message exchanges with consumer actors or in-out (two-way) message exchanges. Replies from consumer actors are mandatory for in-out message exchanges but optional for in-only message exchanges. For replying to a Camel endpoint, the consumer actor uses the very same interface as for replying to any other sender (e.g. to another actor). Examples are self.reply or self.reply_?.

Let's start by defining a simple consumer actor that accepts messages via tcp on port 6200 and replies to the tcp client (tcp support is given by Camel's mina component).

For consumer actors to work, applications need to start a CamelService which is managed by the CamelServiceManager.

When starting a consumer actor, the endpoint defined for that actor will be activated asynchronously by the CamelService. If your application wants to wait for consumer endpoints to be finally activated you can do so with the awaitEndpointActivation method (which is especially useful for testing).

For sending a test message to the consumer actor, the above code uses a Camel ProducerTemplate which can be obtained from the CamelContextManager.

If Camel endpoints, such as the file endpoint, create in-only message exchanges then consumer actors need not reply, by default. The message exchange is completed once the input message has been added to the consumer actor's mailbox.

When placing a file into the data/input directory, the Camel file endpoint will pick up that file and send its content as message to the consumer actor. Once the message is in the actor's mailbox, the file endpoint will delete the corresponding file (see delete=true in the endpoint URI).

If you want to let the consumer actor decide when the file should be deleted, then you'll need to turn auto-acknowledgements off as shown in the following example (autoack = false). In this case the consumer actor must reply with a special Ack message when message processing is done. This asynchronous reply finally causes the file endpoint to delete the consumed file.

Turning auto-acknowledgements on and off is only relevant for in-only message exchanges because, for in-out message exchanges, consumer actors need to reply in any case with an (application-specific) message. Consumer actors may also reply with a Failure message to indicate a processing failure. Failure replies can be made for both in-only and in-out message exchanges. A Failure reply can be done inside receive method but there are better ways as shown in the next sections.

Fault-tolerance and message redelivery

Message processing inside receive may throw exceptions which usually requires a failure response to Camel (i.e. to the consumer endpoint). This is done with a Failure message that contains the failure reason (an instance of Throwable). Instead of catching and handling the exception inside receive, consumer actors should be part of supervisor hierarchies and send failure responses from within restart callback methods. Here's an example of a fault-tolerant file consumer.

The above file consumer overrides the preRestart and postStop callback methods to send reply messages to Camel. A reply within preRestart and postStop is possible after receive has thrown an exception (new feature since Akka 1.1). When receive returns normally it is expected that any necessary reply has already been done within receive.

If the lifecycle of the SupervisedFileConsumer is configured to be PERMANENT, a supervisor will restart the consumer upon failure with a call to preRestart. Within preRestart a Failure reply is sent which causes the file endpoint to redeliver the content of the consumed file and the consumer actor can try to process it again. Should the processing succeed in a second attempt, an Ack is sent within receive. A reply within preRestart must be a safe reply via self.reply_? because an unsafe self.reply will throw an exception when the consumer is restarted without having failed. This can be the case in context of all-for-one restart strategies.
If the lifecycle of the SupervisedFileConsumer is configured to be TEMPORARY, a supervisor will shut down the consumer upon failure with a call to postStop. Within postStop an Ack is sent which causes the file endpoint to delete the file. One can, of course, choose to reply with a Failure here, so that files that couldn't be processed successfully are kept in the input directory. A reply within postStop must be a safe reply via self.reply_? because an unsafe self.reply will throw an exception when the consumer has been stopped by the application (and not by a supervisor) after successful execution of receive.

Another frequently discussed consumer actor example is a fault-tolerant JMS consumer. A JMS consumer actor should acknowledge a message receipt upon successful message processing and trigger a message redelivery on failure. This is exactly the same pattern as described for the SupervisedFileConsumer above. You just need to change the file endpoint URI to a jms or activemq endpoint URI and you're done (of course, you additionally need to configure the JMS connection with a redelivery policy and, optionally, use transacted queues. An explanation how to do this would however exceed the scope of this blog post).

Simplifications and tradeoffs with `blocking=true`

In all the examples so far the internally created Camel routes use the ! (bang) operator to send the input message to the consumer actor. This means that the Camel route does not block a thread waiting for a response. It's an asynchronous reply will cause the Camel route to resume processing. That's also the reason why any exception thrown by receive isn't reported back to Camel directly but must be done explicitly with a Failure response.

If you want that exceptions thrown by receive are reported back to Camel directly (i.e. without sending Failure responses) then you'll need to set blocking = true for the consumer actor. This causes the Camel route to send the input message with the !! (bangbang) operator and to wait for a response. However, this will block a thread until the consumer sends a response or throws an exception within receive. The advantage of this approach is that error handling is strongly simplified in this case but scalability will likely decrease.

Here's an example of a consumer actor that uses the simplified approach to error handling. Any exception thrown by receive will still cause the file endpoint to redeliver the message but a thread will be blocked by Camel during the execution of receive.

No supervisor is needed here. It depends on the non-functional requirements of your application whether to go for this simple but blocking approach or to use a more scalable, non-blocking approach in combination with a supervisor.

Bounded mailboxes and error handling with custom Camel routes

For consumer actors that require a significant amount of time for processing a single message, it can make sense to install a bounded mailbox. A bounded mailbox throws an exception if its capacity is reached and the Camel route tries to add additional messages to the mailbox. Here's an example of a file consumer actor that uses a bounded mailbox with a capacity of 5. Processing is artificially delayed by 1 second using a Thread.sleep.

When, let's say, 10 files are put into the data/input directory, they will be picked up by the file endpoint and added to the actor's mailbox. The capacity of the mailbox will be reached soon because the file endpoint can send messages much faster than the consumer actor can process it. Exceptions thrown by the mailbox are directly reported to the Camel route which causes the file consumer to redeliver messages until they can be added to the mailbox. The same applies to JMS and other endpoints that support redelivery.

When dealing with endpoints that do not support redelivery, one needs to customize the Camel route to the consumer actor with a special error handler that does the redelivery. This is shown for a consumer actor that consumes messages from a direct endpoint.

Here we use onRouteDefinition to define how the Camel route should be customized during its creation. In this example, an error handler is defined that attempts max. 3 redeliveries with a delay of 1000 ms. For details refer to the intercepting route construction section in the akka-camel documentation. When using a producer template to send messages to this endpoint, some of them will be added to the mailbox on first attempt, some of them after a second attempt triggered by the error handler.

The examples presented in this post cover many of the consumer-actor-related questions and topics that have been asked and discussed on the akka-user mailing list. In another post I plan to cover best practices for implementing Akka producer actors.

Akka's grown-up hump

2010-08-30T10:28:00.016+02:00

It's quite some time ago when I last wrote about Akka's Camel integration. An initial version of the akka-camel module was released with Akka 0.7. Meanwhile Akka 0.10 is out with an akka-camel module containing numerous new features and enhancements. Some of them will be briefly described in this blog post.

Java API

The akka-camel module now offers a Java API in addition to the Scala API. Both APIs are fully covered in the online documentation.

Support for typed consumer actors

Methods of typed actors can be published at Camel endpoints by annotating them with @consume. The annotation value defines the endpoint URI. Here's an example of a typed consumer actor in Java.

import org.apache.camel.Body;
import org.apache.camel.Header;
import se.scalablesolutions.akka.actor.TypedActor;
import se.scalablesolutions.akka.camel.consume;

public interface MyTypedConsumer {
   @consume("file:data/foo")
   public void foo(String body);

   @consume("jetty:http://localhost:8877/camel/bar")
   public String bar(@Body String body, @Header("Content-Type") String contentType);
}

public class MyTypedConsumerImpl extends TypedActor implements MyTypedConsumer {
   public void foo(String body) {
       System.out.println(String.format("Received message: ", body));
   }

   public String bar(String body, String contentType) {
       return String.format("body=%s Content-Type header=%s", body, conctentType);
   }
}

When creating an instance of the typed actor with

import se.scalablesolutions.akka.actor.TypedActor;

// Create typed actor and activate endpoints
MyTypedConsumer consumer = TypedActor.newInstance(
   MyTypedConsumer.class, MyTypedConumerImpl.class);

then the actor's foo method can be invoked by dropping a file into the data/foo directory. The file content is passed via the body parameter. The bar method can be invoked by POSTing a message to http://localhost:8877/camel/bar. The HTTP message body is passed via the body parameter and the Content-Type header via the contentType parameter. For parameter binding, Camel's parameter binding annotations are used.

Endpoint lifecycle

Consumer actor endpoints are activated when the actor is started and de-activated when the actor is stopped. This is the case for both typed and untyped actors. An actor can either be stopped explicitly by an application or by a supervisor.

Fault tolerance

When a consumer actor isn't stopped but restarted by a supervisor, the actor's endpoint remains active. Communication partners can continue to exchange messages with the endpoint during the restart phase but message processing will occur only after restart completes. For in-out message exchanges, response times may therefore increase. Communication partners that initiate in-only message exchanges with the endpoint won't see any difference.

Producer actors

Actors that want to produce messages to endpoints either need to mixin the Producer trait (Scala API) or extend the abstract UntypedProducerActor class (Java API). Although the Producer trait was already available in the initial version of akka-camel, many enhancements have been made since then. Most of them are internal enhancements such as performance improvements and support for asynchronous routing. Also, extensions to the API have been made to support

pre-processing of messages before they are sent to an endpoint and
post-processing of messages after they have been received as response from an endpoint.

For example, instead of replying to the original sender (default behavior) a producer actor could do a custom post-processing e.g. by forwarding the response to another actor (together with the initial sender reference)

import se.scalablesolutions.akka.actor.Actor
import se.scalablesolutions.akka.camel.Producer

class MyProducer(target: ActorRef) extends Actor with Producer {
 def endpointUri = "http://example.org/some/external/service"

 override protected def receiveAfterProduce = {
   // do not reply to initial sender but
   // forward result to a target actor
   case msg => target forward msg
 }
}

Forwarding results to other actors makes it easier to create actor-based message processing pipelines that make use of external services. Examples are given in the akka-camel documentation.

Typed actors need to use Camel's ProducerTemplate directly to produce messages to Camel endpoints. A managed instance of a ProducerTemplate can be obtained via CamelContextManager.template.

Asynchronous routing

Since Akka 0.10, Camel's asynchronous routing engine is fully supported: in-out and in-only messages exchanges between endpoints and actors are designed to be asynchronous. This is the case for both, consumer and producer actors.

This is especially important for actors that participate in long-running request-reply interactions with external services. Threads are no longer blocked for the full duration of an in-out message exchange and are available for doing other work. There's also an asynchronous routing example described in the online documentation.

Routes to actors

Typed an untyped actors can also be accessed from Camel routes directly, using Akka's TypedActorComponent and ActorComponent, respectively. These are Camel components supporting typed-actor and and actor endpoint URIs in route definitions. For example,

from("seda:test").to("actor:uuid:12345678");

routes a message from a SEDA queue to an untyped actor with uuid 12345678. The actor endpoint looks up the actor in Akka's actor registry.

The TypedActorComponent is an extension of Camel's bean component where method invocations follow the semantics of the actor model. Here is an example route from a direct endpoint to the foo method of a typed actor.

from("direct:test").to("typed-actor:sample?method=foo");

The typed actor is registered under the name sample in the Camel registry. For more details how to add typed actors to the Camel registry, follow this link.

CamelService

Prerequisite for endpoints being activated when starting consumer actors is a running CamelService. When starting Akka in Kernel mode or using the Akka Initializer in a web application, a CamelService is started automatically. In all other cases a CamelService must be started by the application itself. This can be done either programmatically with

import se.scalablesolutions.akka.camel.CamelServiceManager._

startCamelService

or declaratively in a Spring XML configuration file.

<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xmlns:akka="http://www.akkasource.org/schema/akka"
 xmlns:camel="http://camel.apache.org/schema/spring"
 xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.akkasource.org/schema/akka
http://scalablesolutions.se/akka/akka-0.10.xsd
http://camel.apache.org/schema/spring
http://camel.apache.org/schema/spring/camel-spring.xsd">

 <!– A custom CamelContext (SpringCamelContext) –>
 <camel:camelContext id="camelContext">
   <!– … –>
 </camel:camelContext>

 <!– Create a CamelService using a custom CamelContext –>
 <akka:camel-service>
   <akka:camel-context ref="camelContext" />
 </akka:camel-service>

</beans>

Usage of the element requires the akka-spring jar on the classpath. This example also shows how the Spring-managed CamelService is configured with a custom CamelContext.

A running CamelService can be stopped either by closing the application context or by calling the CamelServiceManager.stopCamelService method.

Outlook

The next Akka release will be Akka 1.0 (targeted for late fall) and akka-camel development will mainly focus on API stabilization. If you'd like to have some additional features in the next Akka release, want to give feedback or ask some questions, please contact the Akka community at the akka-user mailing list.

Akka features for application integration

2010-04-03T10:59:00.009+02:00

Akka is a platform for event-driven, scalable and fault-tolerant architectures on the JVM. It is mainly written in Scala. One of its core features is support for the actor model that provides a higher level of abstraction for writing concurrent and distributed systems.

Since version 0.7, Akka offers a new feature that let actors send and receive messages over a great variety of protocols and APIs. In addition to the native Scala actor API, actors can now exchange messages with other systems over large number of protcols and APIs such as HTTP, SOAP, TCP, FTP, SMTP or JMS, to mention a few. At the moment, approximately 80 protocols and APIs are supported. This new feature is provided by Akka's Camel module.

At the core of this new feature is Apache Camel, IMHO the most powerful and feature-rich integration framework currently available for the JVM. For an introduction to Apache Camel you may want to read this article. Camel comes with a large number of components that provide bindings to different protocols and APIs. Usage of Camel's integration components in Akka is essentially a one-liner. Here's an example.


import se.scalablesolutions.akka.actor.Actor
import se.scalablesolutions.akka.actor.Actor._
import se.scalablesolutions.akka.camel.{Message, Consumer}

class MyActor extends Actor with Consumer {
  def endpointUri =
    "mina:tcp://localhost:6200?textline=true"

  def receive = {
    case msg: Message => { /* ... */}
    case _            => { /* ... */}
  }
}
// start and expose actor via tcp
val myActor = actorOf[MyActor].start

The above example exposes an actor over a tcp endpoint on port 6200 via Apache Camel's Mina component. The endpointUri is an abstract method declared in the Consumer trait. After starting the actor, tcp clients can immediately send messages to and receive responses from that actor. If the message exchange should go over HTTP (via Camel's Jetty component), only the actor's endpointUri must be redefined.


class MyActor extends Actor with Consumer {
  def endpointUri =
    "jetty:http://localhost:8877/example"

  def receive = {
    case msg: Message => { /* ... */}
    case _            => { /* ... */}
  }
}

Actors can also trigger message exchanges with external systems i.e. produce to Camel endpoints.


import se.scalablesolutions.akka.actor.Actor
import se.scalablesolutions.akka.camel.Producer

class MyActor extends Actor with Producer {
  def endpointUri = "jms:queue:example"
  protected def receive = produce
}

In the above example, any message sent to this actor will be added (produced) to the example JMS queue. Producer actors may choose from the same set of Camel components as Consumer actors do.

The number of Camel components is constantly increasing. Akka's Camel module can support these in a plug-and-play manner. Just add them to your application's classpath, define a component-specific endpoint URI and use it to exchange messages over the component-specific protocols or APIs. This is possible because Camel components bind protocol-specific message formats to a Camel-specific normalized message format. The normalized message format hides protocol-specific details from Akka and makes it therefore very easy to support a large number of protocols through a uniform Camel component interface. Akka's Camel module further converts mutable Camel messages into immutable representations which are used by Consumer and Producer actors for pattern matching, transformation, serialization or storage, for example.

Highly-scalable eHealth integration solutions with Akka

One goal I had in mind when implementing the Akka Camel module was to have a basis for building highly-scalable eHealth integration solutions. For eHealth information systems it is becoming increasingly important to support standard interfaces as specified by IHE. Financial support from governments strongly depends on eHealth standard compliance.

Building blocks for implementing standard-compliant eHealth applications are provided by the Open eHealth Integration Platform (IPF). IPF is a mature open source integration platform, based on Apache Camel. It provides, among others, extensive support for IHE actor interfaces. These interfaces are based on Apache Camel's component technology. Therefore, it's a one-liner to expose an Akka actor through an IHE compliant interface. The following example implements the server-side interface of the IHE XDS Registry Stored Query (XDS-ITI18) transaction.


class RSQService extends Actor with Consumer {
  def endpointUri = "xds-iti18:RSQService"

  def receive = {
    case msg: Message => { /* ... */}
    case _            => { /* ... */}
  }
}

In IHE, a message exchange between two participants is called a transaction. Here, the IPF xds-iti18 component is used to implement the server-side interface of the XDS ITI18 transaction. This allows any XDS-ITI18-comaptible client to communicate with the actor over an IHE standard protocol using ebXML/SOAP/HTTP (as defined in the XDS specification). The implementor of the receive method, however, doesn't need to care about all the low-level protocol details (which are scary if you take a closer look). The body of the received message is a high-level object graph containing XDS-ITI18-specific transaction data.

A high-level programming model for implementing eHealth standards is only one of several reasons why I consider Akka as a powerful technical basis for building scalable and fault-tolerant eHealth information systems and integration solutions. Akka's support for NoSQL datastores could further be used for implementing scalabale persistence layers in eHealth applications. Over the next weeks I'm going to explore this field in more detail and will keep you updated with further blog posts on that topic.

Accessing a security-enabled Google App Engine service with Apache Camel

2010-02-07T13:02:00.006+01:00

In a previous post I've described the low-level details for programmatic login to a Google App Engine service from a Java client. Things are getting much easier when using Apache Camel. The recently committed glogin component makes it trivial to login to a remotely deployed Google App Engine service as well as to a local development server. In the following example, an application-specific authorization cookie is obtained with the glogin component. It authorizes a client application to access http://camelcloud.appspot.com on Google App Engine.


import org.apache.camel.Exchange;
import org.apache.camel.ProducerTemplate;
import static org.apache.camel.component.gae.login.GLoginBinding.*;

...

ProducerTemplate template = ...

Exchange result = template.request(
"glogin://camelcloud.appspot.com"
  + "?userName=replaceme@gmail.com"
  + "&password=replaceme", null);
String cookie = result.getOut().getHeader(
  GLOGIN_COOKIE, String.class));

Please note that the password is only sent to the Google Accounts API for authentication. It is never sent to Google App Engine or included into any URL. The obtained authorization cookie is valid for 24 hours and needs to be sent with subsequent requests to the GAE application. If inclusion of user credentials in an endpoint URI is not an option, username and password can also be dynamically set (per request) using Camel message headers:


import org.apache.camel.Exchange;
import org.apache.camel.ProducerTemplate;
import static org.apache.camel.component.gae.login.GLoginBinding.*;

...

ProducerTemplate template = ...

Exchange result = template.request(
"glogin://camelcloud.appspot.com", new Processor() {
  public void process(Exchange exchange) {
  exchange.getIn().setHeader(
    GLOGIN_USER_NAME, "replaceme@gmail.com");
  exchange.getIn().setHeader(
    GLOGIN_PASSWORD, "replaceme");
  }
});
String cookie = result.getOut().getHeader(
  GLOGIN_COOKIE, String.class));

To login to a local development server, the devMode parameter in the endpoint URI must be set to true.


import org.apache.camel.Exchange;
import org.apache.camel.ProducerTemplate;
import static org.apache.camel.component.gae.login.GLoginBinding.*;

...

ProducerTemplate template = ...

Exchange result = template.request(
"glogin://localhost:8888"
   + "?userName=test@example.org"
   + "&devMode=true", null);
String cookie = result.getOut().getHeader(
  GLOGIN_COOKIE, String.class));

The glogin component is part of the Camel Components for Google App Engine.

Add OAuth to your web application with Apache Camel

2010-02-07T11:48:00.007+01:00

OAuth is an open protocol to allow secure API authorization from desktop and web applications. Google, for example, already supports OAuth for authorizing 3rd-party applications to access Google services on behalf of a user.

Recently, I added the gauth component to Apache Camel. You can use it to implement OAuth consumer functionality for any web application with only a few lines of code. gauth endpoints take care of exchanging authorization and access tokens between a web application and an OAuth service provider. At the moment, the gauth component can be used to interact with Google's OAuth services, later versions will support other OAuth providers as well.

From a user's perspective, an example OAuth scenario might look as follows:

The user logs into a web application that uses the Google Calendar API, for example.
To authorize access, the user is redirected to a Google Accounts authorization page where access for the requesting web application can be granted or denied.
After granting access the user is redirected back to the web application and the web application can now access the user's calendar data.
The user can revoke access at any time within Google Accounts.

To implement that scenario with Apache Camel, two routes are needed. The first route obtains an unauthorized request token from Google and then redirects the user to the Google Accounts authorization page:


String encodedCallback = URLEncoder.encode(
 "https://example.org/handler", "UTF-8");
String encodedScope = URLEncoder.encode(
 "http://www.google.com/calendar/feeds/", "UTF-8");

from("jetty:http://0.0.0.0:8080/authorize")
.to("gauth://authorize"
 + "?callback=" + encodedCallback
 + "&scope=" + encodedScope);

In this example, the authorization request is triggered by the user by sending a GET request to http://example.org/authorize (e.g. by clicking a link in the browser). The gauth://authorize endpoint then obtains an unauthorized request token from Google. The scope parameter in the endpoint URI defines which Google service the web application wants to access. After having obtained the token, the endpoint generates a redirect response (302) which redirects the user to the Google Accounts authorization page. After granting access, the user is redirected back to the web application (callback parameter). The callback now contains an authorized request token that must finally be upgraded to an access token. Handling the callback and upgrading to an access token is done in the second route.


from("jetty:https://example.org/handler")
.to("gauth://upgrade")
.to(new StoreTokenProcessor())

The jetty endpoint receives the callback from Google. The gauth://upgrade endpoint takes the authorized request token from the callback and upgrades it to an access token. The route finally stores the long-lived access token for the current user. The next time the user logs into the web application, the access token is already available and the application can continue to access the user's Google Calendar data without needing further user interaction. The user can invalidate the access token at any time within Google Accounts.

Only these two routes are needed to integrate with Google's OAuth provider services. The routes can perfectly co-exist with any other web application framework. Whereas the web framework provides the basis for web application-specific functionality, the OAuth service provider integration is done with Apache Camel. This approach allows for a clean separation of integration logic from application or domain logic.

For handling OAuth requests, web applications can also use other components than Camel's jetty component, such as the servlet component. For adding OAuth to Google App Engine applications, the jetty component needs to be replaced with Camel's ghttp component. Here's an example:


String encodedCallback = URLEncoder.encode(
 "https://camelcloud.appspot.com/handler", "UTF-8");
String encodedScope = URLEncoder.encode(
 "http://www.google.com/calendar/feeds/", "UTF-8");

from("ghttp:///authorize")
.to("gauth://authorize"
 + "?callback=" + encodedCallback
 + "&scope=" + encodedScope);

from("ghttp:///handler")
.to("gauth://upgrade")
.to(new StoreTokenProcessor())

The following figure gives an overview how the OAuth sequence of interactions relate to the gauth://authorize and gauth://upgrade endpoints.

Accessing a Google service with an access token (step 9) is application-specific and not covered by the gauth component. To get access to a user's Google Calendar data with an access token, one could use the GData client library. The gauth component documentation contains an example.

The gauth component is the first step towards a broader support of security standards such as OAuth and OpenID in Apache Camel. I'm currently thinking of the following extensions

A Camel OpenID component
A Camel OpenID/OAuth hybrid component
Support OAuth providers other than Google

The gauth component is currently part of the Camel 2.3 development snapshot (sources).

Accessing a security-enabled Google App Engine service from a Java client

2010-01-14T16:23:00.012+01:00

After a rather long search on Google pages and forums I could only find fragmented information how to programmatically access a Google App Engine service that requires users to authenticate. In this blog post I'm going to summarize my findings for a Java client application.

With programmatic access I mean that the user doesn't need to enter username and password into a login form created by Google but rather into an installed client application and the client coordinates the authentication and authorization process programmatically. The mechanism used here is the ClientLogin for installed applications.

The first step is to obtain an authentication token from the Google Accounts API. The easiest way to do that is with the GData client library for Java.


import java.net.URLEncoder;

import com.google.gdata.client.GoogleAuthTokenFactory;
import com.google.gdata.util.AuthenticationException;

public class AuthExample {

  public static void main(String[] args) throws Exception {

    String username = "myusername@gmail.com";
    String password = "mypassword";
    String serviceName = "ah";

    GoogleAuthTokenFactory factory = new GoogleAuthTokenFactory(serviceName, "", null);
    // Obtain authentication token from Google Accounts
    String token = factory.getAuthToken(username, password, null, null, serviceName, "");

    ...
  }
}

One has to provide username an password and the name of the Google service that should be accessed. For Google App Engine the service name is always ah, regardless of the name of the deployed application. The next step is to do a login at Google App Engine. The login URL is https://example.appspot.com/_ah/login?continue=https%3A%2F%2Fexample.appspot.com%2Fexample&auth=DQAAAJc...qNUA8. The continue query parameter instructs the login service where to rederict after successful login. In this example the redirect goes to https://example.appspot.com/example. The auth query parameter contains the authentication token obtained before.


import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.DefaultHttpClient;

public class AuthExample {

  public static void main(String[] args) throws Exception {
    ...

    String token = ...
    String serviceUrl = "https://example.appspot.com/example";
    String loginUrl = "https://example.appspot.com/_ah/login?continue=" +
    URLEncoder.encode(serviceUrl, "UTF-8") + "&auth=" + token;

    HttpClient httpclient = new DefaultHttpClient();
    HttpGet httpget = new HttpGet(loginUrl);
    HttpResponse response = httpclient.execute(httpget);
    // process response
    // ...

    httpclient.getConnectionManager().shutdown();
  }
}

When the login service sends a redirect after successful login, it also returns a cookie that allows the client to finally access the protected App Engine service at https://example.appspot.com/example. The redirect and cookie handling is done by the httpclient automatically. For the duration of the session the protected App Engine service can be accessed with that cookie.

Update: If the service expects POST requests instead of GET requests then an automated redirect is not an option. In this case, redirect must be disabled for the for the httpclient and a POST request to the serviceUrl must be created manually. Also, the authorization cookie must be set explicitly.


import org.apache.http.Header;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.params.ClientPNames;
import org.apache.http.impl.client.DefaultHttpClient;

public class AuthExample {

  public static void main(String[] args) throws Exception {
    ...

    String token = ...
    String loginUrl = "https://example.appspot.com/_ah/login?auth=" + token;
    String serviceUrl = "https://example.appspot.com/example";

    HttpClient httpclient = new DefaultHttpClient();
    httpclient.getParams().setBooleanParameter(ClientPNames.HANDLE_REDIRECTS, false);
    HttpGet httpget = new HttpGet(loginUrl);
    HttpResponse response = httpclient.execute(httpget);
    // Get cookie returned from login service
    Header[] headers = response.getHeaders("Set-Cookie");
    httpclient.getConnectionManager().shutdown();   

    httpclient = new DefaultHttpClient();
    HttpPost httppost = new HttpPost(serviceUrl);
    // set cookie returned by login service
    for (Header header : headers) {
      httppost.addHeader("Cookie", header.getValue());
    }
    // set request entity body
    // ...

    response = httpclient.execute(httppost);
    // process response
    // ...

    httpclient.getConnectionManager().shutdown();   
  }
}

Update: Login to a local development server. To get access to a security-enabled application on the local development server there's no need for getting an authentication token. Instead, POST an email address and a redirect URL to http://localhost:<port>/_ah/login and the server returns an authorization cookie. Here's an example:

HttpClient httpClient = new DefaultHttpClient();
httpClient.getParams().setBooleanParameter(
  ClientPNames.HANDLE_REDIRECTS, false);
// POST login data to GAE SDK dev server
HttpPost httpPost = new HttpPost(
  "http://localhost:8888/_ah/login");
httpPost.setHeader("Content-Type",
  "application/x-www-form-urlencoded");
String email = URLEncoder.encode(
  "test@example.com", "UTF-8");
String redirectUrl = URLEncoder.encode(
  "http://localhost:8888", "UTF-8");
httpPost.setEntity(new StringEntity(
  "email=" + email + "&continue=" + redirectUrl));
HttpResponse response = httpClient.execute(httpPost);
// Extract authorization cookie from response
String cookie = response.getFirstHeader("Set-Cookie").getValue();
httpClient.getConnectionManager().shutdown();
// Create a new client and access the secured
// service with the authorization cookie
httpClient = new DefaultHttpClient();
HttpGet httpget = new HttpGet("http://localhost:8888");
httpget.addHeader("Cookie", cookie);
response = httpClient.execute(httpget);
System.out.println(IOUtils.toString(response.getEntity().getContent()));
httpClient.getConnectionManager().shutdown();

New features in grails-jaxrs 0.3

2009-12-12T08:52:00.006+01:00

In this blog post I present some new features of the recently released grails-jaxrs 0.3 plugin. A complete list of new features is available in the release notes. A feature overview and links to the complete documentation is on the plugin home page.

grails-jaxrs is a Grails plugin that supports the development of RESTful web services based on the Java API for RESTful Web Services (JSR 311: JAX-RS). It is targeted at developers who want to structure the web service layer of an application in a JSR 311 compatible way but still want to continue to use Grails' powerful features such as GORM, automated XML and JSON marshalling, Grails services, Grails filters and so on. This plugin is an alternative to Grails' built-in mechanism for implementing RESTful web services.

The following example shows how to do content negotiation for Grails domain objects. Grails domain classes like


class Person {
  String firstName
  String lastName
}

can now be used in JAX-RS resource methods directly (e.g. Person parameter in the create method):


import static javax.ws.rs.core.UriBuilder.fromPath

import javax.ws.rs.Consumes
import javax.ws.rs.Path
import javax.ws.rs.Produces
import javax.ws.rs.POST
import javax.ws.rs.core.Response

@Path('/api/person')
@Consumes(['application/xml','application/json'])
@Produces(['application/xml','application/json'])
class PersonCollectionResource {

  @POST
  Response create(Person person) {
      person.save() // use GORM
      URI uri = fromPath(person.id as String).build()
      Response.created(uri).entity(person).build()
  }

  // ...
       
}

Content negotiation and conversion between domain objects and their XML or JSON representations is done by domain object providers. There's no need any more for application code to deal with representation formats directly.

The PersonCollectionResource.create method handles POST requests for creating new Person objects in the database. The method uses GORM to persist the domain object. Clients can send either XML or JSON representations for POSTing person data (see Content-Type header):


POST /hello/api/person HTTP/1.1
Content-Type: application/xml
Accept: application/xml
Host: localhost:8080
Content-Length: 78

<person>
<firstname>Sam</firstname>
<lastname>Hill</lastname>
</person>


POST /hello/api/person HTTP/1.1
Content-Type: application/json
Accept: application/json
Host: localhost:8080
Content-Length: 58

{"class":"Person","firstName":"Fabien","lastName":"Barel"}

In either case, the plugin will convert it to a Person object, as required by the person parameter. For creating a response the method uses the JAX-RS API. It first creates a URI for the response Location header and uses the Response builder to set the status code to 201 (created) and the response entity. Note that the method itself doesn't create an XML or JSON representation of the response domain object. This is again done by a domain object provider which uses the Accept request header to determine the response representation format. The responses to the above POST requests are:


HTTP/1.1 201 Created
Content-Type: application/xml
Location: http://localhost:8080/hello/api/person/1
Transfer-Encoding: chunked
Server: Jetty(6.1.14)

<?xml version="1.0" encoding="UTF-8"?>
<person id="1">
<firstname>Sam</firstname>
<lastname>Hill</lastname>
</person>

and


HTTP/1.1 201 Created
Content-Type: application/json
Location: http://localhost:8080/hello/api/person/2
Transfer-Encoding: chunked
Server: Jetty(6.1.14)

{"class":"Person","id":"2","firstName":"Fabien","lastName":"Barel"}

The PersonCollectionResource.create method is even more verbose than necessary. It could equally be written as


import static org.grails.jaxrs.response.Responses.*

@Path('/api/person')
@Consumes(['application/xml','application/json'])
@Produces(['application/xml','application/json'])
class PersonCollectionResource {

  @POST
  Response create(Person person) {
      created person.save()
  }

  // ...
       
}

using helper methods (a mini-DSL) from org.grails.jaxrs.response.Responses. That's exactly the code that is generated when using scaffolding for the Person domain class i.e.


grails generate-resources person

With the grails-jaxrs scaffolding feature, one can generate RESTful service interface for domain objects supporting the HTTP methods POST, GET, PUT and DELETE. A scaffolding example is given in the Scaffolding section of the grails-jaxrs documentation, a walk through the generated code is in the Using GORM section.

By default, grails-jaxrs uses Grail's XML and JSON converters for converting between domain objects and their XML or JSON representations. Applications can easily customize this conversion logic as explained in the Custom entity providers section.

Besides usage of GORM, grails-jaxrs also supports auto-injection of Grails services into JAX-RS resource and provider classes or usage of Grails filters, to mention a few. With version 0.3 the included JAX-RS implementations have been upgraded to their latest versions: Jersey 1.1.4.1 and Restlet 2.0-M6.

Camel components for Google App Engine

2009-11-17T06:27:00.002+01:00

The upcoming Apache Camel version 2.1 will include components for connecting to the cloud computing services of Google App Engine. At the moment the following three components are available.

ghttp: Provides connectivity to the GAE URL fetch service but can also be used to receive messages from servlets
gtask: Supports asynchronous message processing on GAE by using the task queueing service as message queue.
gmail: Supports sending of emails via the GAE mail service. Receiving mails is not supported yet but will be added later.

Camel components for the other Google App Engine cloud computing services such as Memcache service, XMPP service, Images service, Datastore Service and the Authentication service are planned.

There's also a tutorial that explains how to develop a non-trivial Camel GAE application using the Camel components for GAE.

From a conceptual point of view, connecting to cloud computing services via Camel components introduces an abstraction-layer that decouples Camel applications from provider-specific cloud service interfaces. Supporting several cloud computing environments in Camel can significantly reduce the burden of migrating Camel applications from one provider to another. The Camel components for Google App Engine are a first step into this direction.

First steps with Apache Camel on Google App Engine

2009-10-18T09:32:00.007+02:00

This post describes how to get a simple Camel 2 application running on Google App Engine (GAE). I'll focus on the workarounds and fixes that were necessary to succeed. Please note that the following descriptions are by no means best-practices or recommendations. They only describe my first steps for which better solutions will likely exist in the future. I plan to work on improvements to

make Camel deployments on GAE easier and to
allow Camel applications access GAE services via Camel components

For my experiments, I was using a Camel 2.1 development snapshot, the App Engine SDK 1.2.6 and the Google Plugin for Eclipse which makes local testing and remote deployment very easy. The Camel components I used are:

camel-core
camel-spring
camel-servlet
camel-http

The following snippet shows the route definition of the sample application. It uses the camel-servlet component to receive input via HTTP, converts the HTTP request body to a String, prepends a "Hello " to the body and returns the result.


package example;

import org.apache.camel.builder.RouteBuilder;

public class ExampleRoute extends RouteBuilder {

  @Override
  public void configure() throws Exception {
     from("servlet:/test")
     .convertBodyTo(String.class)
     .transform(constant("Hello ").append(body()));
  }
}

The route doesn't make use of any GAE services (URL fetch, tasks queues, storage, mail ...) Also, message processing is synchronous because GAE doesn't allow applications to create their own threads. For example, using SEDA or JMS queues will not work.

For processing HTTP requests, I created my own servlet class and extended the CamelHttpTransportServlet from the camel-servlet component.


package example;

import org.apache.camel.component.servlet.CamelHttpTransportServlet;
import org.apache.camel.management.JmxSystemPropertyKeys;

public class ExampleServlet extends CamelHttpTransportServlet {

 static {
     System.setProperty(JmxSystemPropertyKeys.DISABLED, "true");
 }

}

The only thing this servlet does is to disable all JMX-related functionality because the GAE JRE doesn't support JMX. All request processing and dispatching is done by the CamelHttpTransportServlet. Configuring the servlet in the web.xml was done as follows.


<servlet>
<servlet-name>CamelServlet</servlet-name>
<servlet-class>example.ExampleServlet</servlet-class>
<init-param>
   <param-name>contextConfigLocation</param-name>
   <param-value>context.xml</param-value>
</init-param>
</servlet>

<servlet-mapping>
<servlet-name>CamelServlet</servlet-name>
<url-pattern>/camel/*</url-pattern>
</servlet-mapping>

The servlet init-param points to the Spring application context that configures the route builder and the Camel context:


<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">

<bean id="camelContext"
   class="org.apache.camel.spring.CamelContextFactoryBean">
   <property name="builderRefs">
       <list>
           <ref bean="routeBuilderRef"/>
       </list>
   </property>
</bean>

<bean id="routeBuilderRef"
   class="org.apache.camel.model.RouteBuilderDefinition">
   <constructor-arg value="routeBuilder" />
</bean>

<bean id="routeBuilder"
   class="example.ExampleRoute">
</bean>

</beans>

A severe limitation is that one cannot use the Camel-specific configuration XML schema from the http://camel.apache.org/schema/spring namespace for configuring the Camel context. The problem is that the CamelNamespaceHandler uses JAXB to parse bean definitions which isn't supported by GAE either. One has to fallback to plain old <bean> definitions (POBD?) to configure the Camel context in Spring. Using Spring JavaConfig or something similar would make more sense here but I didn't try it.

Another JAXB-releated problem arises with Camel's Spring DSL. It is also processed with JAXB and therefore cannot be used on GAE.

Going completely without Spring leads to another problem. In this case the CamelContext uses a JndiRegistry by default that depends on javax.naming.InitialContext. This class isn't on the JRE whitelist either. Writing a simple Map-based implementation of org.apache.camel.impl.Registry and configuring the CamelContext with it does the trick.

The last obstacle to get the sample application running was to replace the Camel's UuidGenerator with another one that uses java.util.UUID from the JRE. Camel's original UuidGenerator also uses a class that is not on the JRE whitelist. Since replacement by configuration was not possible, changes to the Camel code base were necessary (patch already submitted).

After deploying the application to GAE and POSTing a request containing "Martin" to http://<appname>.appspot.com/camel/test I was able to send myself greetings. In the URL, <appname> must of course be replaced with the name of an existing application.

IPF 2.0 milestone 2 and IPF Tools milestones released

2009-10-14T15:00:00.003+02:00

I'm pleased to announce the following milestone releases from the IPF core project and the IPF Tools project.

IPF 2.0 milestone 2 (release notes)
IPF Tools
- IPF Runtime 2.0 milestone 2 (release notes)
- IPF Manager 2.0 milestone 2 (release notes)
- IPF IDE 1.0 milestone 2 (release notes)

IPF 2.0 milestone 2

This release is feature-equivalent to IPF 1.7.0 but runs on Camel 2.0. Users who plan to upgrade to IPF 2.0 or Camel 2.0 in the near future are highly recommended to use this milestone release. Please note that IPF 2.0 is not backwards-compatible to IPF 1.x. This is mainly due to non-backwards compatible API changes in Camel 2.0. It is therefore important to carefully read the Camel 2.0.0 release notes as well as the IPF 2.0-m2 upgrade notes.

With the release of IPF 2.0-m2 and IPF 1.7.0, IPF 1.x development will go into maintainance mode and new features will be developed on the IPF 2.0 development branch. We leave it open whether to backport selected IPF 2.x features to IPF 1.x. Please add any backport requests to the IPF issue tracker.

Other changes compared to IPF 1.7.0 are:

The platform manager has been moved to the IPF Tools project.
The IPF OSGi distributable (IPF runtime) and the IPF OSGi documentation have been moved to the IPF Tools project.
The HL7-independent parts of the mapping service have been factored out into a new commons-map component.
The IPF 2.0 documentation has been forked from the IPF 1.7 documentation and revised.

IPF Runtime 2.0 milestone 2

The IPF Runtime is an IPF distribution that is running on the Equinox OSGi platform. It is available as Eclipse plugin or as standalone package. The runtime is used to develop OSGi-based IPF applications.

IPF Manager 2.0 milestone 2

IPF Manager is an Eclipse application for managing IPF services and applications. It is available as Eclipse plugin or as standalone package. In its current state it provides a flow management user interface and a general-purpose JMX client. The IPF Manager is compatible with IPF Runtime 2.0-m2.

IPF IDE 1.0 milestone 2

The IPF IDE supports developers in creating, testing and packaging IPF applications within the Eclipse plugin development environment (PDE) on top of the IPF runtime. The IPF IDE is compatible with IPF Runtime 2.0-m2.

IPF 1.7.0 released

2009-10-07T11:09:00.003+02:00

I'm pleased to announce the release of IPF 1.7.0. The main focus of this release was support for clinical standards, in particular the IHE profiles XDS.a, XDS.b, PIX, PDQ and support for the clinical document architecture (CDA) and the Continuity of Care Document (CCD) content profile. The release notes are here.

With IPF's IHE support, IHE actor interfaces can be implemented in IPF routes via URIs. This is as simple as using other Camel or IPF components such as the HTTP or JMS components. The URIs denote individual transactions (ITI) in IHE profiles. For example

from('xds-iti18:myIti18Service')
...

implements the 'XDS Registry Stored Query' service interface of an XDS document registry and can be used from any XDS ITI18-compliant consumer. Such a consumer can also be implemented using the same IPF xds-iti18 component on client side e.g.

...
.to('xds-iti18://somehost:8080/myWebApp/services/myIti18Service')

All the low-level details like communicating with ebXML messages over SOAP etc. is handled by that component. IPF routes deal with easy-to-use object representations of messages exchanged within IHE transactions. The full list of supported transactions is given in the IHE quick reference.

With IPF's CDA and CCD support clinical documents can be created, parsed, rendered and queried/analyzed using a domain-specific language (DSL). This DSL hides away most of the technical details you usually encounter when dealing with the complex XML-representation of clinical documents. In addition to these content-DSL extensions, IPF also provides some route DSL extension for parsing, validating and marshalling CDA documents in IPF routes.

Here's an excerpt of new IPF 1.7.0 features added since 1.6.0

* IHE support
** IHE XDS.a+b transactions (ITI 14-18, 41-43)
** IHE PIX transactions (ITI 8-10)
** IHE PDQ transactions (ITI 21-22)
** IHE ATNA for all the above transactions
* CDA support
** Generic CDA support
** CCD profile support
* Advanced XML processing
** Caching XSLT transmogrifier
** Schematron validator
* Detailed XDS tutorial
* Performance measurement support
* Scheduled flow management database cleanup
* ...

IPF 1.7.0 is based on Camel 1.6. In parallel, a Camel 2.0-based version is developed on the IPF 2.0 branch. The current development snapshot is feature-equivalent with IPF 1.7.0 but runs on Camel 2.0. The next IPF 2.0 milestone 2 release will follow within the next one or two weeks. I recommend you to use the 2.0 milestone releases unless you upgrade from an older IPF 1.x release. After releasing IPF 2.0.0, work on IPF 1.x will go into maintainance mode (but we leave it open whether to backport selected IPF 2.x features).

Exciting new features in IPF 2.0 which didn't make it into IPF 1.7 are Eclipse-based IPF development tools and improvements to IPF's OSGi support. These features are developed and documented in a separate IPF Tools project. The Eclipse-based IPF management client has been moved to this project as well. I'll give a more detailed IPF 2.0 overview in a separate post.

Many thanks to the whole development team and contributors for their excellent and high-quality work!

New release of JSR 311 plugin for Grails

2009-09-23T07:29:00.006+02:00

It's about two weeks ago that the latest version (0.2) of the JSR 311 plugin for Grails (grails-jaxrs) has been released. In contrast to the first release (0.1) , which was mainly a proof-of-concept, this new release focused, among other things, on a closer Grails integration. In particular

JAX-RS classes like resource classes and entity providers are now auto-detected by the plugin. There's no need any more to add them to the Spring application context manually.

JAX-RS classes managed by the plugin can be changed at runtime in development mode. Code changes are detected by the plugin and reloaded in the same way as Grails controllers or services are.

Services and other Spring beans are auto-injected by-name into JAX-RS resources and providers.

The plugin also extends Grails' command-line interface for creating JAX-RS resources from scratch as well as generating JAX-RS resources from existing domain objects (scaffolding). With scaffolding RESTful service interfaces can be auto-generated for individual domain objects (still early-access).

Antoher enhancement in this version is support for deployments on the Google App Engine (GAE). This required to support Restlet as JAX-RS implementation in addition to Jersey. In contrast to Jersey, Restlet can be deployed to GAE and since version 2.0-m4 its JAX-RS extension as well. For a running example go to http://grails-jaxrs.appspot.com/test?name=World. Please note that initializing Grails applications on GAE can take very long (up tp 30 seconds) at the moment. Subsequent requests are served much faster, of course.

Also the documentation has been and extended and completely revised. The easiest way to get started with the plugin is the Getting Started guide.

JSR 311 plugin for Grails

2009-07-08T10:45:00.005+02:00

Project homepage: grails-jaxrs

Recently, I wanted to make a full two-weeks mountain bike tour through the Alps but bad weather on a few days caused me to make detours on the less rocky roads of JSR 311 (JAX-RS: The Java API for RESTful Web Services). I took the chance reading the spec, did some hacking with Jersey and I must really say I like to work with it. I liked it so much that I started to use it inside Grails to combine it with features such as GORM and XML and JSON marshalling.

From my experiments I factored out what I think is reusable into a Grails plugin named grails-jaxrs and made it open source. The plugin takes care of initializing Jersey inside a Grails application, implements a controller that does the dispatch to Jersey and provides some Grails-specific entity provider implementations.

Here's a very simple example of a resource class that represents a collection of notes and that makes use of Grails object relational mapping (GORM) and XML marshaling:


import grails.converters.*

import javax.ws.rs.Consumes
import javax.ws.rs.GET
import javax.ws.rs.Path
import javax.ws.rs.PathParam
import javax.ws.rs.POST
import javax.ws.rs.Produces
import javax.ws.rs.core.Response
import javax.ws.rs.core.UriBuilder

@Path('/notes')
class NotesResource {

   @POST
   @Consumes('text/plain')
   @Produces('text/xml')
   Response addNote(String text) {

       // Create new Note object and store save it to DB
       def note = new Note(text:text).save()
     
       // Construct the URI for the newly created note
       URI uri = UriBuilder.fromPath(note.id as String).build()
     
       // Return an XML representation of the note object
       // along with a Location response header with the URI
       Response.created(uri).entity(note as XML).build()
   }

   @GET
   @Produces('text/xml')
   Response getNotes() {

       // Find all notes in the database and return
       // an XML representation of the note list
       Response.ok(Note.findAll() as XML).build()
   }

}

A note is an object that contains some text. Notes are instances of the Note Grails domain class. A collection of notes are made available through a RESTful service interface using the above NotesResource class and the grails-jaxrs plugin. A POST to http://host:port/notes/ creates a new note. A GET to http://host:port/notes returns an XML repsresentation of all existing notes in the database. For more details refer to the getting started tutorial.

Moving towards IPF 1.7

2009-05-24T10:43:00.003+02:00

Work on the next IPF release is in progress and focuses on

Components for implementing IHE actor interfaces (XDS.a, XDS.b, PIX, PDQ). Implementing IHE actor interfaces will be as simple as using other Camel components for communication (e.g. the HTTP component). IPF IHE components represent transactions in IHE profiles. For example, to create a web service for the server-side of the ITI-41 transaction (provide and register document set) from the XDS.b profile just write from('xdsb-iti41:service1') in your route definition. xdsb-iti41 is the name of the component, service1 the endpoint name. The rest of the route has to deal with connecting to (proprietary) backend systems that implement the corresponding actor functionality (document registry/repository). For more detailed information refer to this article (section Outlook).

DSL (Groovy builder) for creating CDA documents. This DSL supports the creation of structurally correct CDA documents by enforcing CDA-relevant schema definitions but without dealing with low-level XML details. See also outlook on IPF's CDA support.

Extension of OSGi support. Not all features of IPF have been fully OSGi-enabled with release 1.6, like the large binary support or the event infrastructure. This will be fixed with IPF 1.7. You'll also be able to use the new IHE components on OSGi platforms. We also plan to extend existing Camel components to make use of standard OSGi services such as the HTTP service.

Better IDE (Eclipse) integration, especially for developing IPF OSGi applications. This includes application development and packaging with Eclipse PDE tools and providing an Eclipse update site for downloading the IPF runtime. Code completion for DSL extensions is currently under discussion.

Performance testing framework. DSL extensions to performance-measure IPF applications including calculation and reporting of message processing statistics during load tests.

In parallel we will also create a SVN branch for experimenting with Camel 2.0 milestone releases. I'll let you know about our upgrade/migration experiences in a separate blog post.

Open eHealth Integration Platform 1.6.0 released

2009-04-13T17:55:00.001+02:00

Last week we've released the Open eHealth Integration Platform (IPF) version 1.6.0. IPF is an extension of the Apache Camel routing and mediation engine. It has an application programming layer based on the Groovy programming language and comes with comprehensive support for message processing and connecting systems in the eHealth domain. Please see the release notes for further details.

Give it a try! We welcome your feedback!

Martin Krasser's Blog

Building an Event-Sourced Web Application - Part 2: Projections, Persistence, Consumers and Web Interface

Overview

Running the example application

Service Layer Enhancements

Event Logs

CQRS and Consistency

Business Processes

Web Interface

Building an Event-Sourced Web Application - Part 1: Domain Model, Events and State

Application Overview

Example application

Domain Model

Service Layer

Summary

Akka Producer Actors: New Features and Best Practices

Basic usage

Failures

Forwarding results

Correlation identifiers

Fault-tolerance

Akka Consumer Actors: New Features and Best Practices

Basic usage

Fault-tolerance and message redelivery

Simplifications and tradeoffs with blocking=true

Bounded mailboxes and error handling with custom Camel routes

Akka's grown-up hump

Akka features for application integration

Highly-scalable eHealth integration solutions with Akka

Accessing a security-enabled Google App Engine service with Apache Camel

Add OAuth to your web application with Apache Camel

Accessing a security-enabled Google App Engine service from a Java client

New features in grails-jaxrs 0.3

Camel components for Google App Engine

First steps with Apache Camel on Google App Engine

IPF 2.0 milestone 2 and IPF Tools milestones released

IPF 1.7.0 released

New release of JSR 311 plugin for Grails

JSR 311 plugin for Grails

Moving towards IPF 1.7

Open eHealth Integration Platform 1.6.0 released

Simplifications and tradeoffs with `blocking=true`