This package contains an implementation of a simple filesystem-based message queue system.

Introduction

The model of the message queueing system is as follows:

applications (including map-reduce jobs) may create their own separate message queueing area. Alternatively, they can specifically ask for a named message queue, belonging to a different application or existing as a system-wide queue. Message queues are created under "/mq" and then the message queue id (for map-reduce jobs this is a job id, or it can be any other name passed as job id to the constructor).
Please see the example for more information.
a single unit of information passing through queues is a {@link org.apache.nutch.util.msg.Msg}, which has a unique identifier (consisting of creation time and publisher name), string subject, and content ({@link org.apache.hadoop.io.Writable}).
single {@link org.apache.nutch.util.msg.MsgQueue} in fact consists of any number of topics. There are four predefined ones: in, out, err, and ctrl.
messages are published to topics, which present a sequential view of messages, sorted by msgId (which corresponds to their order of arrival).each message queue may periodically poll for changes ({@link org.apache.nutch.util.msg.MsgQueue#startPolling()}), using a separate thread. Polling updates the list of topics and messages. Poll interval is configurable, and defaults to 5 sec.
each detected change in the queue (add/remove topic, add/remove message) may be communicated to registered listeners. Out-of-band messages are not supported in this version, but it's not too complicated to add them. Applications can create listeners watching queues for newly added messages, or deleted messages, added topics or deleted topics, etc.
each instance of {@link org.apache.nutch.util.msg.MsgQueue} using the same physical queue maintains its own view of the queue, keeping track of topics and messages that it considers "processed and discarded". In other words, multiple readers and creators may modify queues, and each knows which messages it already processed and which ones are new. In a similar fashion, instances may willfully "remove" certain topics from their view, even though these topics still physically exist and are available for other instances (and later on they can "add" them to their view again).
This somewhat complicated feature was implemented in order to support multiple readers for the same message (e.g. many tasks per one mapred job). Each task needs to register for the same queue, and if they didn't have their own views of the queue, messages would be consumed by the first task that got to them. As it is implemented now, each task may consume messages at its own pace. At the end of the job applications may elect to keep the queue around or to destroy it (and thus remove all topics and messages in it).
messages, topics and queues may be destroyed by any user, at which point they are physically removed from the filesystem. All users will gradually update their views, during the next poll operation.
there is a command-line tool to examine and modify queues, and also to retrieve and send simple text messages. You can run it like this:
```
   bin/nutch org.apache.nutch.util.msg.MsgQueueTool ...many options...
```

Example

Here's an example of how a map-reduce {@link org.apache.hadoop.mapred.Mapper} implementation could use message queues for inter-task communication, specifically to gracefully stop or resume the processing:

public class MQAwareJob implements Mapper {
	static final int JOB_RUN   = 0;
	static final int JOB_PAUSE = 1;
	static final int JOB_STOP  = 2;
	...
	private MsgQueue mq;
	private int runState;
	...

	private static class CtrlListener implements MsgQueueListener {
		private MQAwareJob job;
		
		public CtrlListener(MQAwareJob job) {
			this.job = job;
		}
		
		public void processEvent(MsgTopic topic, MsgTopicEvent evt) throws Exception {
			if (evt.getType() == evt.TYPE_MSG_ADDED) {
				Msg m = topic.getMsg(evt.getMsgId());
				if (m.getSubject().equals("STOP")) {
					job.runState = JOB_STOP;
				} else if (m.getSubject().equals("PAUSE")) {
					job.runState = JOB_PAUSE;
				} else if (m.getSubject().equals("START")) {
					job.runState = JOB_START;
				}
			}
		}
	}			
		
	public void configure(JobConf conf) {
		runState = JOB_RUN;
		CtrlListener ctrlListener = new CtrlListener(this);
		// create or connect to a message queue for this job
		mq = new MsgQueue(conf);
		mq.getMsgTopic(MsgTopic.TOPIC_CONTROL).addMsgQueueListener(ctrlListener);
		// register any other queue or topic listeners, as needed
		...
	}

	public void map( ... ) {
		switch (runState) {
		case JOB_STOP:
			// discard all further input
			return;
		case JOB_PAUSE:
			int cnt = 0;
			while (runState == JOB_PAUSE) {
				try {
					Thread.sleep(1000);
				} catch (Exception e) {};
				reporter.setStatus("paused " + cnt++ + "s");
			}
			break;
		default:
			// just run
		}
		...
	}
}