20.5.13

GMaill Accessing Using JavaMail


<util:properties id="gmailProperties587">
<prop key="mail.smtp.host">smtp.gmail.com</prop>
<prop key="mail.smtp.auth">true</prop>
<prop key="mail.smtp.port">587</prop>
<prop key="mail.debug">true</prop>
<prop key="mail.smtp.starttls.enable">true</prop>
<prop key="mail.smtp.quitwait">false</prop>
</util:properties>

<util:properties id="gmailProperties465">
<prop key="mail.smtp.host">smtp.gmail.com</prop>
<prop key="mail.smtp.auth">true</prop>
<prop key="mail.smtp.port">465</prop>
<prop key="mail.debug">true</prop>
<prop key="mail.smtp.quitwait">false</prop>
<prop key="mail.smtp.socketFactory.port">465</prop>
<prop key="mail.smtp.socketFactory.class">javax.net.ssl.SSLSocketFactory</prop>
<prop key="mail.smtp.socketFactory.fallback">false</prop>
</util:properties>

15.5.13

Service Activator in Spring Integration

Travel of Software Developer: Understanding Service Activator

Just want to continue this topic and focus on Spring Integration's implementation.

How Spring Integration Describes its Service Activator?


  • A Service Activator is a generic endpoint for connecting a service instance to the messaging system.
That's great!
  • A generic endpoint.
  • Connecting a service instance.
  • To the messaging system.

A Generic Endpoint

As I mentioned in the previous post, Service Activator is usually abused for its generality. Does this confirm this problem here? I am not sure.
  • The input Message Channel must be configured, and if the service method to be invoked is capable of returning a value, an output Message Channel may also be provided.
    • The output channel is optional, since each Message may also provide its own "Return Address" header. This same rule applies for all consumer endpoints.
This matches the Pattern characteristics in EIP:
  • A Service Activator can be one-way (request only) or two-way (Request-Reply).
  • The service can be as simple as a method call - synchronous and non-remote - perhaps part of a Service Layer.

More Information about Service Activator in Spring Integration

Implementation of Service Activator in Spring Integration


This is the Factory Bean used to create the Service Activator bean. The implementation of Service Activator's Factory Bean is very much different from the way GatewayProxyFactoryBean, but I don't know why.
GatewayProxyFactoryBean comes all the way from IntegrationObjectSupport, AbstractEndpoint, and AbstractPollingEndpoint, which makes sense for a Messaging Endpoint. While Service Activator is also a Messaging Endpoint, it derives from AbstractSimpleMessageHandlerFactoryBean<H>, then AbstractStandardMessageHandlerFactoryBean. It also makes from the Factory Bean's point of view. And the real handler for Service Activator is: ServiceActivatingHandler:

Although Service Activator can also use SpEL:
  • Since Spring Integration 2.0, Service Activators can also benefit from SpEL.
I am going to focus only on ServiceActivatingHandler now.

ServiceActivatingHandler

@Override
protected Object handleRequestMessage(Message<?> message) {
try {
return this.processor.processMessage(message);
}
catch (Exception e) {
if (e instanceof RuntimeException) {
throw (RuntimeException) e;
}
throw new MessageHandlingException(message, "failure occurred in Service Activator '" + this + "'", e);
}
}

This is how ServiceActivatingHandler is implemented. But wait, I don't see Asynchronous!! And MessageProcessor<T> sounds new to me now. Let's take a look at how this MessageProcessor is constructed.

MessageProcessor<T> and MethodInvokingMessageProcessor<T>

@Override
MessageHandler createMethodInvokingHandler(Object targetObject, String targetMethodName) {
ServiceActivatingHandler handler = (StringUtils.hasText(targetMethodName))
? new ServiceActivatingHandler(targetObject, targetMethodName)
: new ServiceActivatingHandler(targetObject);
return this.configureHandler(handler);
}

So, this is as simple as a constructor.

public ServiceActivatingHandler(final Object object) {
this(new MethodInvokingMessageProcessor<Object>(object, ServiceActivator.class));
}

And the essential part is MethodInvokingMessageProcessor<Object>, while in Gateway, this is MethodInvocationGateway extends MessagingGatewaySupport.

private final MessagingMethodInvokerHelper<T> delegate;

public T processMessage(Message<?> message) {
try {
return delegate.process(message);
}
catch (Exception e) {
throw new MessageHandlingException(message, e);
}
}

MessagingMethodInvokerHelper<T>, The Delegate

I don't want to look into this class right now, because I believe it just try to find the method and call the right method, the valid basic method reflection. If I find I am wrong later, I will come back.

@Async and Annotation Driven Task Executor

  • To enable both @Scheduled and @Async annotations, simply include the 'annotation-driven' element from the task namespace in your configuration.
This is the way to support Asynchronous Service Activator with Direct Channel.

Implementation of Service Activator with @Async 


Message Return from ServiceActivator with @Async

In the above example, I used a one-way ServiceActivator. This is a very natural way for using Service Activator. You can inject the successive channel into ServiceActivator, so that ServiceActivator could send out messages or receive messages flexibly. I think this is the so-called Half-sync/Half-Async way, not sure though.

However, sometimes we might want to get some output from ServiceActivator. In another word, we want a two-way ServiceActivator. You won't have problem with Synchronous Service Activator, but you will with @Async. If you do it in a normal way, such as:

@Async
@ServiceActivator
public Message<Object> handleMessage(Message<String> msg)

You won't get anything as its output. Why?

AnnotationAsyncExecutionInterceptor.invoke(MethodInvocation)

public Object invoke(final MethodInvocation invocation) throws Throwable {
Future<?> result = this.determineAsyncExecutor(invocation.getMethod()).submit(
new Callable<Object>() {
public Object call() throws Exception {
try {
Object result = invocation.proceed();
if (result instanceof Future) {
return ((Future<?>) result).get();
}
}
catch (Throwable ex) {
ReflectionUtils.rethrowException(ex);
}
return null;
}
});
if (Future.class.isAssignableFrom(invocation.getMethod().getReturnType())) {
return result;
}
else {
return null;
}
}

This is the place @Async is handled. And you see, only if the methods return type is an instance of Future<T>, the result will be kept. Otherwise, the result will simply be discarded.

@Async
@ServiceActivator
public Future<Message<Object>> handleMessage(Message<String> msg) throws Exception {
return new AsyncResult<Message<Object>>(XXXX);
}

For a complete example, please see this.

But how can Service Activator pass down the return to successive handlers? I don't think there is an easy solution. And that's maybe why you won't see any document from Spring Integration telling you to use @Async for Service Activator.

Service Activator with QueueChannel




Asynchronous Gateway and Parallel DirectChannel

Parallel DirectChannel

DirectChannel could work in a Parallel way:

<int:channel id="channel-in">
<int:dispatcher load-balancer="round-robin" task-executor="taskExec"/>
</int:channel>
However, a channel with a dispatcher doesn't necessary be handled in parallel. In the above example, if the Gateway expects some return, DirectChannel won't returned immediately after execute the job in another executor. Although the execution is driven into another thread, but the current thread will be blocked for the reply:

MessagingTemplate:

private <S, R> Message<R> doSendAndReceive(MessageChannel channel, Message<S> requestMessage) {
Object originalReplyChannelHeader = requestMessage.getHeaders().getReplyChannel();
Object originalErrorChannelHeader = requestMessage.getHeaders().getErrorChannel();
TemporaryReplyChannel replyChannel = new TemporaryReplyChannel(this.receiveTimeout);
requestMessage = MessageBuilder.fromMessage(requestMessage)
.setReplyChannel(replyChannel)
.setErrorChannel(replyChannel)
.build();
this.doSend(channel, requestMessage);
Message<R> reply = this.doReceive(replyChannel);
if (reply != null) {
reply = MessageBuilder.fromMessage(reply)
.setHeader(MessageHeaders.REPLY_CHANNEL, originalReplyChannelHeader)
.setHeader(MessageHeaders.ERROR_CHANNEL, originalErrorChannelHeader)
.build();
}
return reply;
}

The doReceived(...) method will blocked the caller's thread until it receives the reply from the replyChannel, which technically makes the work in sync.
If you really want to have some reply from the DirectChannel while you want to have it work in parallel, you need Asynchronous Gateway or something similar.

Asynchronous Gateway

You can of course use the default asynchronous executor, which is AsyncTaskExecutor. But the default executor will create as many threads as it can to run the Gateway. Sometimes it will drain the resources of your system without proper control. So you might want to manually set up an executor with some control.




Revert Change in Eclipse/Git

http://stackoverflow.com/questions/6788881/undo-single-file-local-uncommitted-change-in-egit-e-g-svn-revert

Google Code, Git, and Eclipse

Create a new Project in Google Code

  • Open any projects.
  • From "My favorites" drop-down menu, you will see "Create a project...", click on it.
  • Fill in all Options except Project labels. Choose Git as the Version Control System.
  • Click on "Create project" button.

Find Access Information

  • Open "Source" tab.
  • etc. etc. etc.

Clone a Repository

  • Run this command
git clone https://<your user name>@code.google.com/p/simple-demo-set/ 

  • Create a RADME.txt under the repository.
  • git add README.txt
  • git commit -m "Add the first file"
  • create $HOME/.netrc (I'm using cygwin)
machine code.google.com
login XXXX
password GOOGLEGENERATED
  • git push -u origin master

Add Git Repository into Eclipse

  • Open Git Repositories Exploring Perspective
  • Click on "Add an existing local Git Repository to this view"
  • Select the designated directory location and clikc on Finish.

Share Project from Eclipse

  • Choose the repository instead of create a local one.

Create .gitignore

It seems that Eclipse's global ignorance settings don't work for Git. You have to create your own .gitignore for the git repository. These are the entry you probably want to add.

.classpath
.project
.settings
target


14.5.13

Storm?





  • Whereas Hadoop targets batch processing, Storm is an always-active service that receives and processes unbound streams of data.
    • Like Hadoop, Storm is a distributed system that offers massive scalability for applications that store and manipulate big data. 
    • Unlike Hadoop, it delivers that data instantaneously, in realtime.
  • It is written primarily in Clojure and supports Java by default.
  • Use cases
    • Realtime analytics
    • Online machine learning
    • Continuous computation
    • Distributed RPC
    • ETL
  • How does storm differ from Hadoop?
    • The simple answer is that Storm analyzes realtime data while Hadoop analyze offline data.
    • In truth, the two frameworks complement one another more than they compete.
  • Hadoop
    • Provides its own file system (HDFS)
    • Manages both data and code/tasks.
    • It divides data into blocks and when a "job" executes, it pushes analysis code close to the data it is analyzing.
      • This is how Hadoop avoids the overhead of network communication in loading data -- keeping the analysis code next to the data enables Hadoop to read it faster by orders of magnitude.
    • MapReduce
      • Hadoop partitions data into chunks and passes those chunks to mappers that map keys to values.
      • Reducers then assemble those mapped key/value pairs into a usable output.
      • The MapReduce paradigm operates quite elegantly but is targeted at data analysis.
    • HDFS
      • In order to leverage all the power of Hadoop application data must be stored in the HDFS file system.
  • Storm
    • Storm solves a different problem altogether.
    • Realtime
      • meaning right now
      • Storm is interested in understanding things that are happening in realtime, and interpreting them.
    • File System
      • Storm does not have its own file system.
    • Programming Paradigm
      • Its programming paradigm is quite a bit different from Hadoop's.
      • Storm is all about obtaining chunks of data, known as spouts, from somewhere and passing that data through various processing components, known as bolts.
      • Storm's data processing mechanism is extremely fast and is meant to help you identify live trends as they are happening.
      • Unlike Hadoop, Storm doesn't care what happened yesterday or last week.
  • Architecture
    • At the highest level, Storm is comprised of topologies.
      • A topology is a graph of computations
        • Each node contains processing logic and each path between nodes indicates how data should be passed between nodes.
    • In side of topologies you have networks of streams, which are unbounded sequences of tuples.
      • Storm provides a mechanism to transform streams into new streams using spouts and bolts.
      • Spouts
        • Spouts generate streams, which can pull data from a site like Twitter of Facebook and then publish it in an abstract format.
      • Bolts
        • Bolts consume input streams, process them, and then optionally generate new streams.
    • Tuples
      • Storm's data model is represented by tuples.
        • A tuples is a named list of values of any type.
        • Storm supports all primititve types, Strings, and byte-arrays and you can build your own serializer if you want to use your own object types.
      • Your spouts will "emit" tuples 
      • And your bolts will consume them.
      • Your bolts may also emit tuples if their output is destined to be processed by another bolt downstream.
      • Basically, emitting tuples is the mechanism for passing data from a spout to a bolt, or from a bolt to another bolt.
This is quite a normal architecture for expandable or scalable computatal system, or network. And basically, the name, Topologies, the graph, makes a lot of sense.

It's an Oriented Graph. If we add a root node above Spouts, it would look like a tree but with shared children. It reminds me the Traveling Saleman Problem (TSP, Travelling salesman problem - Wikipedia, the free encyclopedia).
  • Storm Cluster
    • A Storm Cluster is somewhat similar to Hadoop clusters, but while a Hadoop cluster runs map-reduce jobs, Storm runs topologies.
      • Map-reduce jobs eventually end.
      • Topologies are destined to run until you explicitly kill them.
    • Storm clusters define two types of nodes
      • Master Node
        • This node runs a daemon process called Nimbus.
        • Nimbus is responsible for distributing code across the cluster, assigning tasks to machines, and monitoring the success and failure of units of work.
      • Worker Nodes
        • These nodes run a daemon process called the Supervisor.
        • A Supervisor is responsible for listening for work assignments for its machine.
        • It then subsequently starts and stops worker processes. Each worker process executes a subset of a topology, so that the execution of a topology is spread across a multitude of worker processes running on a multitude of machines.
This is a typical cluster architecture. I used to designed a system with the Master Node named MC (MultiController), and the Worker Nodes named LC (LocalController). MC takes care of job distribution. But MC doesn't monitor the success or failure of units of work. It monitors the LC's health and relocate the Work Node to ensure High Availability.

My system was not designed for dynamic expansion. There are only three fixed layers. LC, the Worker Node runs as a daemon and supervises a list of Engines. Those Engines are created dynamically by some configuration package, the work assignments if you will. It starts and stops these Engines, the Worker Processes if you will. Those Worker Processes work basically independently, which made my cluster simpler then this. I think having each worker process executes a subset of a topology, and coordinates those topology will be a challenge.

In my cluster, each Worker Nodes takes care of the success or failure of its own Worker Processes.
  • ZooKeeper
    • Sitting between the Nimbus and the various Supervisors is ZooKeeper.
    • ZooKeeper's goal is to enable highly reliable distributed coordination, mainly by acting as a centralized service for distributed cluster functionality.
    • Storm topologies are deployed to the Nimbus and then the Nimbus deploys spouts and bolts to Supervisors.
      • When it comes time to execute spouts and bolts, the Nimbus communicates with the Supervisors by passing messages to ZooKeepers.
      • Zookeepers maintain all state for the topology, which allows the Nimbus and Supervisors to be fail-fast and stateless: if the Nimbus or Supervisor processes go down then the state of processing is not lost;
      • If the Nimbus or Supervisor processes go down then the state of processing is not lost;
      • Work is reassigned to another Supervisor and processing continues.
      • Then, when the Nimbus or a Supervisor is restarted, they simply rejoin the cluster and their capacity is added to the cluster.
My first question would be: how many Nimbus in the system? It seems only one. If this is true, we need somewhere to eliminate this single-failure-point (Single point of failure - Wikipedia, the free encyclopedia). So I think Zookeeper actions as an independent configuration storage out of Nimbus. Nimbus delegates Zookeepers to deployed Supervisors and manage them. Zookeepers are clustered and do not have single point of failure problem.

Of course, this is what I guess so far.

  • Storm Development Environment
    • Local Mode:
      • Storm executes topologies completely in-process by simulating worker nodes using threads.
    • Distributed Mode:
      • In distributed mode, it runs across a cluster of machines.