Index: ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsBackendQueueProcessor.java
===================================================================
--- ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsBackendQueueProcessor.java	Mon Jun 22 22:39:19 CEST 2009
+++ ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsBackendQueueProcessor.java	Mon Jun 22 22:39:19 CEST 2009
@@ -0,0 +1,77 @@
+package org.hibernate.search.backend.impl.jgroups;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.jgroups.ChannelClosedException;
+import org.jgroups.ChannelNotConnectedException;
+import org.jgroups.Message;
+import org.slf4j.Logger;
+
+import org.hibernate.search.SearchException;
+import org.hibernate.search.backend.LuceneWork;
+import org.hibernate.search.backend.OptimizeLuceneWork;
+import org.hibernate.search.util.LoggerFactory;
+
+/**
+ * Responsible for sending Lucene works from slave nodes to master node
+ *
+ * @author Lukasz Moren
+ */
+public class JGroupsBackendQueueProcessor implements Runnable {
+
+	protected static final Logger log = LoggerFactory.make();
+
+	private final JGroupsBackendQueueProcessorFactory factory;
+	private final List<LuceneWork> queue;
+
+	public JGroupsBackendQueueProcessor(List<LuceneWork> queue, JGroupsBackendQueueProcessorFactory factory) {
+		this.factory = factory;
+		this.queue = queue;
+	}
+
+	@SuppressWarnings("unchecked")
+	public void run() {
+		List<LuceneWork> filteredQueue = new ArrayList<LuceneWork>( queue );
+		log.trace( "Preparing {} Lucene works to be sent to master node.", filteredQueue.size() );
+
+		for ( LuceneWork work : queue ) {
+			if ( work instanceof OptimizeLuceneWork ) {
+				//we don't want optimization to be propagated
+				filteredQueue.remove( work );
+			}
+		}
+		log.trace(
+				"Filtering: optimized Lucene works are not going to be sent to master node. There is {} Lucene works after filtering.",
+				filteredQueue.size()
+		);
+		if ( filteredQueue.size() == 0 ) {
+			log.trace(
+					"Nothing to send. Propagating works to a cluster has been skipped."
+			);
+			return;
+		}
+
+		/* Creates and send message with lucene works to master.
+		 * As long as message destination address is null, Lucene works will be received by all listeners that implements
+		 * org.jgroups.MessageListener interface, multiple master nodes in cluster are allowed. */
+		try {
+			Message message = new Message( null, factory.getAddress(), ( Serializable ) filteredQueue );
+			factory.getChannel().send( message );
+			log.trace(
+					"Lucene works have been sent from slave {} to master node.", factory.getAddress()
+			);
+		}
+		catch ( ChannelNotConnectedException e ) {
+			throw new SearchException(
+					"Unable to send Lucene work. Channel is not connected to: "
+							+ factory.getClusterName()
+			);
+		}
+		catch ( ChannelClosedException e ) {
+			throw new SearchException( "Unable to send Lucene work. Attempt to send message on closed JGroups channel" );
+		}
+
+	}
+}
Index: ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/master/TShirt.java
===================================================================
--- ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/master/TShirt.java	Wed Jun 10 00:09:04 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/master/TShirt.java	Wed Jun 10 00:09:04 CEST 2009
@@ -0,0 +1,50 @@
+//$Id: TShirt.java 11625 2007-06-04 16:21:54Z epbernard $
+package org.hibernate.search.test.jgroups.master;
+
+import org.hibernate.search.annotations.DocumentId;
+import org.hibernate.search.annotations.Field;
+import org.hibernate.search.annotations.Index;
+import org.hibernate.search.annotations.Indexed;
+
+import javax.persistence.Entity;
+import javax.persistence.GeneratedValue;
+import javax.persistence.Id;
+
+/**
+ * @author Emmanuel Bernard
+ */
+@Entity
+@Indexed
+public class TShirt {
+	@Id
+	@GeneratedValue
+	@DocumentId
+	private int id;
+	@Field(index= Index.TOKENIZED)
+	private String logo;
+	private String size;
+
+	public int getId() {
+		return id;
+	}
+
+	public void setId(int id) {
+		this.id = id;
+	}
+
+	public String getLogo() {
+		return logo;
+	}
+
+	public void setLogo(String logo) {
+		this.logo = logo;
+	}
+
+	public String getSize() {
+		return size;
+	}
+
+	public void setSize(String size) {
+		this.size = size;
+	}
+}
\ No newline at end of file
Index: ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/MasterJGroupsBackendQueueProcessorFactory.java
===================================================================
--- ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/MasterJGroupsBackendQueueProcessorFactory.java	Tue Jun 23 21:02:29 CEST 2009
+++ ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/MasterJGroupsBackendQueueProcessorFactory.java	Tue Jun 23 21:02:29 CEST 2009
@@ -0,0 +1,58 @@
+package org.hibernate.search.backend.impl.jgroups;
+
+import java.util.List;
+import java.util.Properties;
+
+import org.jgroups.Receiver;
+
+import org.hibernate.search.backend.LuceneWork;
+import org.hibernate.search.backend.impl.lucene.LuceneBackendQueueProcessorFactory;
+import org.hibernate.search.engine.SearchFactoryImplementor;
+
+/**
+ * Backend factory used in JGroups clustering mode in master node.
+ * Wraps {@link LuceneBackendQueueProcessorFactory} with providing extra
+ * functionality to receive Lucene works from slave nodes.
+ *
+ * @author Lukasz Moren
+ * @see org.hibernate.search.backend.impl.lucene.LuceneBackendQueueProcessorFactory
+ * @see org.hibernate.search.backend.impl.jgroups.SlaveJGroupsBackendQueueProcessorFactory
+ */
+public class MasterJGroupsBackendQueueProcessorFactory extends JGroupsBackendQueueProcessorFactory {
+
+	private LuceneBackendQueueProcessorFactory luceneBackendQueueProcessorFactory;
+	private Receiver masterListener;
+
+	@Override
+	public void initialize(Properties props, SearchFactoryImplementor searchFactory) {
+		super.initialize( props, searchFactory );
+		initLuceneBackendQueueProcessorFactory( props, searchFactory );
+
+		registerMasterListener();
+	}
+
+	public Runnable getProcessor(List<LuceneWork> queue) {
+		return luceneBackendQueueProcessorFactory.getProcessor( queue );
+	}
+
+	private void registerMasterListener() {
+		//register JGroups receiver in master node to get Lucene docs from slave nodes
+		masterListener = new JGroupsMasterMessageListener( searchFactory );
+		channel.setReceiver( masterListener );
+	}
+
+	private void initLuceneBackendQueueProcessorFactory(Properties props, SearchFactoryImplementor searchFactory) {
+		luceneBackendQueueProcessorFactory = new LuceneBackendQueueProcessorFactory();
+		luceneBackendQueueProcessorFactory.initialize( props, searchFactory );
+	}
+
+	public Receiver getMasterListener() {
+		return masterListener;
+	}
+
+	@Override
+	public void close() {
+		super.close();
+		luceneBackendQueueProcessorFactory.close();
+	}
+}
Index: ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsBackendQueueProcessorFactory.java
===================================================================
--- ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsBackendQueueProcessorFactory.java	Mon Jun 22 23:36:31 CEST 2009
+++ ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsBackendQueueProcessorFactory.java	Mon Jun 22 23:36:31 CEST 2009
@@ -0,0 +1,181 @@
+package org.hibernate.search.backend.impl.jgroups;
+
+import java.net.URL;
+import java.util.List;
+import java.util.Properties;
+
+import org.jgroups.Address;
+import org.jgroups.Channel;
+import org.jgroups.ChannelException;
+import org.jgroups.JChannel;
+import org.slf4j.Logger;
+
+import org.hibernate.search.Environment;
+import org.hibernate.search.SearchException;
+import org.hibernate.search.backend.BackendQueueProcessorFactory;
+import org.hibernate.search.backend.LuceneWork;
+import org.hibernate.search.engine.SearchFactoryImplementor;
+import org.hibernate.search.util.LoggerFactory;
+import org.hibernate.search.util.XMLHelper;
+import org.hibernate.util.ConfigHelper;
+
+
+/**
+ * Common base class for Master and Slave BackendQueueProcessorFactories
+ *
+ * @author Lukasz Moren
+ */
+public abstract class JGroupsBackendQueueProcessorFactory implements BackendQueueProcessorFactory {
+
+	protected static final Logger log = LoggerFactory.make();
+
+	public static final String JGROUPS_PREFIX = Environment.WORKER_BACKEND + ".jgroups.";
+
+	public static final String CONFIGURATION_STRING = JGROUPS_PREFIX + "configurationString";
+	public static final String CONFIGURATION_XML = JGROUPS_PREFIX + "configurationXml";
+	public static final String CONFIGURATION_FILE = JGROUPS_PREFIX + "configurationFile";
+	private static final String DEFAULT_JGROUPS_CONFIGURATION_FILE = "flush-udp.xml";
+
+	public static final String JG_CLUSTER_NAME = JGROUPS_PREFIX + "clusterName";
+
+	protected String clusterName = "HSearchCluster";
+	protected SearchFactoryImplementor searchFactory;
+	protected Channel channel = null;
+	protected Address address;
+
+	public void initialize(Properties props, SearchFactoryImplementor searchFactory) {
+		this.searchFactory = searchFactory;
+
+		if ( props.containsKey( JG_CLUSTER_NAME ) ) {
+			setClusterName( props.getProperty( JG_CLUSTER_NAME ) );
+		}
+
+		prepareJGroupsChannel( props );
+	}
+
+	private void prepareJGroupsChannel(Properties props) {
+		log.info( "Starting JGroups Channel" );
+		try {
+			buildChannel( props );
+			channel.setOpt( Channel.AUTO_RECONNECT, Boolean.TRUE );
+			channel.connect( clusterName );
+		}
+		catch ( ChannelException e ) {
+			throw new SearchException( "Unable to connect to: [" + clusterName + "] JGroups channel" );
+		}
+		log.info( "Connected to cluster [ {} ]. The node address is {}", clusterName, getAddress() );
+
+		if ( !channel.flushSupported() ) {
+			log.warn(
+					"FLUSH is not present in your JGroups stack!  FLUSH is needed to ensure messages are not dropped while new nodes join the cluster.  Will proceed, but inconsistencies may arise!"
+			);
+		}
+	}
+
+	/**
+	 * Reads congiguration and builds channnel with its base.
+	 * In order of preference - we first look for an external JGroups file, then a set of XML properties, and
+	 * finally the legacy JGroups String properties.
+	 *
+	 * @param props configuratuion file
+	 */
+	private void buildChannel(Properties props) {
+		String cfg;
+		if ( props != null ) {
+			if ( props.containsKey( CONFIGURATION_FILE ) ) {
+				cfg = props.getProperty( CONFIGURATION_FILE );
+				try {
+					channel = new JChannel( ConfigHelper.locateConfig( cfg ) );
+				}
+				catch ( Exception e ) {
+					log.error( "Error while trying to create a channel using config files: {}", cfg );
+					throw new SearchException( e );
+				}
+			}
+
+			if ( props.containsKey( CONFIGURATION_XML ) ) {
+				cfg = props.getProperty( CONFIGURATION_XML );
+				try {
+					channel = new JChannel( XMLHelper.elementFromString( cfg ) );
+				}
+				catch ( Exception e ) {
+					log.error( "Error while trying to create a channel using config XML: {}", cfg );
+					throw new SearchException( e );
+				}
+			}
+
+			if ( props.containsKey( CONFIGURATION_STRING ) ) {
+				cfg = props.getProperty( CONFIGURATION_STRING );
+				try {
+					channel = new JChannel( cfg );
+				}
+				catch ( Exception e ) {
+					log.error( "Error while trying to create a channel using config string: {}", cfg );
+					throw new SearchException( e );
+				}
+			}
+		}
+
+		if ( channel == null ) {
+			log.info(
+					"Unable to use any JGroups configuration mechanisms provided in properties {}.  Using default JGroups configuration file!",
+					props
+			);
+			try {
+				URL fileUrl = ConfigHelper.locateConfig( DEFAULT_JGROUPS_CONFIGURATION_FILE );
+				if ( fileUrl != null ) {
+					channel = new JChannel( fileUrl );
+				}
+				else {
+					log.warn(
+							"Default JGroups configuration file was not found. Attempt to start JGroups channel with default configuration!"
+					);
+					channel = new JChannel();
+				}
+			}
+			catch ( ChannelException e ) {
+				throw new SearchException( "Unable to start JGroups channel", e );
+			}
+		}
+	}
+
+	public abstract Runnable getProcessor(List<LuceneWork> queue);
+
+	public void close() {
+		try {
+			if ( channel != null && channel.isOpen() ) {
+				log.info( "Disconnecting and closing JGroups Channel" );
+				channel.disconnect();
+				channel.close();
+			}
+		}
+		catch ( Exception toLog ) {
+			log.error( "Problem closing channel; setting it to null", toLog );
+			channel = null;
+		}
+	}
+
+	public Channel getChannel() {
+		return channel;
+	}
+
+	public void setClusterName(String clusterName) {
+		this.clusterName = clusterName;
+	}
+
+	public String getClusterName() {
+		return clusterName;
+	}
+
+	/**
+	 * Cluster's node address
+	 *
+	 * @return Address
+	 */
+	public Address getAddress() {
+		if ( address == null && channel != null ) {
+			address = channel.getLocalAddress();
+		}
+		return address;
+	}
+}
Index: ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/common/MultipleSessionsSearchTestCase.java
===================================================================
--- ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/common/MultipleSessionsSearchTestCase.java	Sun Jun 21 16:47:54 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/common/MultipleSessionsSearchTestCase.java	Sun Jun 21 16:47:54 CEST 2009
@@ -0,0 +1,150 @@
+package org.hibernate.search.test.jgroups.common;
+
+import java.io.InputStream;
+
+import org.slf4j.Logger;
+
+import org.hibernate.SessionFactory;
+import org.hibernate.cfg.AnnotationConfiguration;
+import org.hibernate.cfg.Configuration;
+import org.hibernate.classic.Session;
+import org.hibernate.dialect.Dialect;
+import org.hibernate.search.test.SearchTestCase;
+import org.hibernate.search.util.FileHelper;
+
+/**
+ * Test class to simulate clustered environment (one master, and one slave node)
+ *
+ * @author Lukasz Moren
+ */
+public abstract class MultipleSessionsSearchTestCase extends SearchTestCase {
+
+	private static final Logger log = org.hibernate.search.util.LoggerFactory.make();
+
+	private String masterCopy = "/master/copy";
+
+	/**
+	 * The lucene index directory which is specific to the master node.
+	 */
+	private String masterMain = "/master/main";
+
+	/**
+	 * The lucene index directory which is specific to the slave node.
+	 */
+	private String slave = "/slave";
+
+
+	protected static SessionFactory slaveSessionFactory;
+
+	/**
+	 * Common configuration for all slave nodes
+	 */
+	private Configuration commonCfg;
+
+	@Override
+	protected void configure(Configuration cfg) {
+		super.configure( cfg );
+
+		//master
+		cfg.setProperty( "hibernate.search.default.sourceBase", getIndexDir().getAbsolutePath() + masterCopy );
+		cfg.setProperty( "hibernate.search.default.indexBase", getIndexDir().getAbsolutePath() + masterMain );
+		cfg.setProperty( "hibernate.search.default.refresh", "1" );
+		cfg.setProperty(
+				"hibernate.search.default.directory_provider", "org.hibernate.search.store.FSMasterDirectoryProvider"
+		);
+	}
+
+	protected void commonConfigure(Configuration cfg) {
+		super.configure( cfg );
+
+		//slave(s)
+		cfg.setProperty( "hibernate.search.default.sourceBase", getIndexDir().getAbsolutePath() + masterCopy );
+		cfg.setProperty( "hibernate.search.default.indexBase", getIndexDir().getAbsolutePath() + slave );
+		cfg.setProperty( "hibernate.search.default.refresh", "1" );
+		cfg.setProperty(
+				"hibernate.search.default.directory_provider", "org.hibernate.search.store.FSSlaveDirectoryProvider"
+		);
+	}
+
+	@Override
+	protected void setUp() throws Exception {
+		if ( getIndexDir().exists() ) {
+			FileHelper.delete( getIndexDir() );
+		}
+		super.setUp();
+		buildCommonSessionFactory( getCommonMappings(), getCommonAnnotatedPackages(), getCommonXmlFiles() );
+	}
+
+	@Override
+	protected void tearDown() throws Exception {
+		super.tearDown();
+
+		//close session factories and clean index files
+		if ( slaveSessionFactory != null ) {
+			slaveSessionFactory.close();
+		}
+		if ( getSessions() != null ) {
+			getSessions().close();
+		}
+		log.info( "Deleting test directory {} ", getIndexDir().getAbsolutePath() );
+		FileHelper.delete( getIndexDir() );
+	}
+
+	private void buildCommonSessionFactory(Class<?>[] classes, String[] packages, String[] xmlFiles) throws Exception {
+		try {
+			if ( getSlaveSessionFactory() != null ) {
+				getSlaveSessionFactory().close();
+			}
+
+			setCommonCfg( new AnnotationConfiguration() );
+			commonConfigure( commonCfg );
+			if ( recreateSchema() ) {
+				commonCfg.setProperty( org.hibernate.cfg.Environment.HBM2DDL_AUTO, "create-drop" );
+			}
+			for ( String aPackage : packages ) {
+				( ( AnnotationConfiguration ) getCommonConfiguration() ).addPackage( aPackage );
+			}
+			for ( Class<?> aClass : classes ) {
+				( ( AnnotationConfiguration ) getCommonConfiguration() ).addAnnotatedClass( aClass );
+			}
+			for ( String xmlFile : xmlFiles ) {
+				InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream( xmlFile );
+				getCommonConfiguration().addInputStream( is );
+			}
+			setDialect( Dialect.getDialect() );
+			slaveSessionFactory = getCommonConfiguration().buildSessionFactory();
+		}
+		catch ( Exception e ) {
+			e.printStackTrace();
+			throw e;
+		}
+	}
+
+	private void setCommonCfg(Configuration configuration) {
+		this.commonCfg = configuration;
+	}
+
+	protected Configuration getCommonConfiguration() {
+		return commonCfg;
+	}
+
+	protected Session getSlaveSession() {
+		return slaveSessionFactory.openSession();
+	}
+
+	protected static SessionFactory getSlaveSessionFactory() {
+		return slaveSessionFactory;
+	}
+
+	private String[] getCommonAnnotatedPackages() {
+		return new String[] { };
+	}
+
+	private String[] getCommonXmlFiles() {
+		return new String[] { };
+	}
+
+	protected abstract Class<?>[] getMappings();
+
+	protected abstract Class<?>[] getCommonMappings();
+}
Index: ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/common/JGroupsCommonTest.java
===================================================================
--- ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/common/JGroupsCommonTest.java	Mon Jun 22 23:35:26 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/common/JGroupsCommonTest.java	Mon Jun 22 23:35:26 CEST 2009
@@ -0,0 +1,128 @@
+package org.hibernate.search.test.jgroups.common;
+
+import java.util.List;
+
+import org.apache.lucene.analysis.StopAnalyzer;
+import org.apache.lucene.queryParser.QueryParser;
+import org.apache.lucene.search.Query;
+
+import org.hibernate.HibernateException;
+import org.hibernate.Session;
+import org.hibernate.Transaction;
+import org.hibernate.cfg.Configuration;
+import org.hibernate.search.Environment;
+import org.hibernate.search.FullTextSession;
+import org.hibernate.search.Search;
+import org.hibernate.search.backend.impl.jgroups.JGroupsBackendQueueProcessorFactory;
+import org.hibernate.search.test.jgroups.master.TShirt;
+
+/**
+ * In case of running test outside Hibernate Search Maven configuration set following VM configuration:
+ * <br><br>
+ * <code>
+ * 	-Djava.net.preferIPv4Stack=true -Djgroups.bind_addr=127.0.0.1
+ * </code>
+ * @author Lukasz Moren
+ */
+
+public class JGroupsCommonTest extends MultipleSessionsSearchTestCase {
+
+	public static final String CHANNEL_NAME = "jgroups_test_channel";
+	private static final String DEFAULT_JGROUPS_CONFIGURATION_FILE = "flush-udp.xml";
+
+	public void testJGroupsBackend() throws Exception {
+
+		//get slave session
+		Session s = getSlaveSession();
+		Transaction tx = s.beginTransaction();
+		TShirt ts = new TShirt();
+		ts.setLogo( "Boston" );
+		ts.setSize( "XXL" );
+		TShirt ts2 = new TShirt();
+		ts2.setLogo( "Mapple leaves" );
+		ts2.setSize( "L" );
+		s.persist( ts );
+		s.persist( ts2 );
+		tx.commit();
+
+		Thread.sleep( 3000 );
+
+		FullTextSession ftSess = Search.getFullTextSession( openSession() );
+		ftSess.getTransaction().begin();
+		QueryParser parser = new QueryParser( "id", new StopAnalyzer() );
+		Query luceneQuery = parser.parse( "logo:Boston or logo:Mapple leaves" );
+		org.hibernate.Query query = ftSess.createFullTextQuery( luceneQuery );
+		List result = query.list();
+
+		assertEquals( 2, result.size() );
+
+		s = getSlaveSession();
+		tx = s.beginTransaction();
+		ts = ( TShirt ) s.get( TShirt.class, ts.getId() );
+		ts.setLogo( "Peter pan" );
+		tx.commit();
+
+		//need to sleep for the message consumption
+		Thread.sleep( 3000 );
+
+		parser = new QueryParser( "id", new StopAnalyzer() );
+		luceneQuery = parser.parse( "logo:Peter pan" );
+		query = ftSess.createFullTextQuery( luceneQuery );
+		result = query.list();
+		assertEquals( 1, result.size() );
+
+		s = getSlaveSession();
+		tx = s.beginTransaction();
+		s.delete( s.get( TShirt.class, ts.getId() ) );
+		s.delete( s.get( TShirt.class, ts2.getId() ) );
+		tx.commit();
+
+		//Need to sleep for the message consumption
+		Thread.sleep( 3000 );
+
+		parser = new QueryParser( "id", new StopAnalyzer() );
+		luceneQuery = parser.parse( "logo:Boston or logo:Mapple leaves" );
+		query = ftSess.createFullTextQuery( luceneQuery );
+		result = query.list();
+		assertEquals( 0, result.size() );
+
+		ftSess.close();
+		s.close();
+
+	}
+
+	@Override
+	protected void configure(Configuration cfg) {
+		//master jgroups configuration
+		super.configure( cfg );
+		cfg.setProperty( Environment.WORKER_BACKEND, "jgroupsMaster" );
+		cfg.setProperty( JGroupsBackendQueueProcessorFactory.CONFIGURATION_FILE, DEFAULT_JGROUPS_CONFIGURATION_FILE );
+	}
+
+	@Override
+	protected void commonConfigure(Configuration cfg) {
+		//slave jgroups configuration
+		super.commonConfigure( cfg );
+		cfg.setProperty( Environment.WORKER_BACKEND, "jgroupsSlave" );
+		cfg.setProperty( JGroupsBackendQueueProcessorFactory.CONFIGURATION_FILE, DEFAULT_JGROUPS_CONFIGURATION_FILE );
+	}
+
+	public static Session getSession() throws HibernateException {
+		return sessions.openSession();
+	}
+
+	@Override
+	protected Class<?>[] getMappings() {
+		return new Class[] {
+				TShirt.class
+		};
+	}
+
+	protected Class<?>[] getCommonMappings() {
+		return new Class[] {
+				TShirt.class
+		};
+	}
+
+
+}
Index: ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/slave/JGroupsSlaveTest.java
===================================================================
--- ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/slave/JGroupsSlaveTest.java	Tue Jun 23 09:16:45 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/slave/JGroupsSlaveTest.java	Tue Jun 23 09:16:45 CEST 2009
@@ -0,0 +1,164 @@
+package org.hibernate.search.test.jgroups.slave;
+
+import org.jgroups.Channel;
+import org.jgroups.JChannel;
+
+import org.hibernate.Session;
+import org.hibernate.Transaction;
+import org.hibernate.cfg.Configuration;
+import org.hibernate.search.Environment;
+import org.hibernate.search.backend.impl.jgroups.JGroupsBackendQueueProcessorFactory;
+import org.hibernate.search.test.SearchTestCase;
+import org.hibernate.search.util.XMLHelper;
+
+/**
+ * Tests that the Slave node in a JGroups cluster can properly send messages to the channel.
+ * <p/>
+ * In case of running test outside Hibernate Search Maven configuration set following VM configuration:
+ * <br><br>
+ * <code>
+ * -Djava.net.preferIPv4Stack=true -Djgroups.bind_addr=127.0.0.1
+ * </code>
+ *
+ * @author Lukasz Moren
+ */
+
+public class JGroupsSlaveTest extends SearchTestCase {
+
+	public static final String CHANNEL_NAME = "HSearchCluster";
+
+	private Channel channel;
+
+	public void testMessageSend() throws Exception {
+
+		JGroupsReceiver.reset();
+
+		Session s = openSession();
+		Transaction tx = s.beginTransaction();
+		TShirt ts = new TShirt();
+		ts.setLogo( "Boston" );
+		ts.setSize( "XXL" );
+		TShirt ts2 = new TShirt();
+		ts2.setLogo( "Mapple leaves" );
+		ts2.setSize( "L" );
+		s.persist( ts );
+		s.persist( ts2 );
+		tx.commit();
+
+		//need to sleep for the message consumption
+		Thread.sleep( 500 );
+
+		assertEquals( 1, JGroupsReceiver.queues );
+		assertEquals( 2, JGroupsReceiver.works );
+
+		JGroupsReceiver.reset();
+		s = openSession();
+		tx = s.beginTransaction();
+		ts = ( TShirt ) s.get( TShirt.class, ts.getId() );
+		ts.setLogo( "Peter pan" );
+		tx.commit();
+
+		//need to sleep for the message consumption
+		Thread.sleep( 500 );
+
+		assertEquals( 1, JGroupsReceiver.queues );
+		assertEquals( 2, JGroupsReceiver.works );
+
+		JGroupsReceiver.reset();
+		s = openSession();
+		tx = s.beginTransaction();
+		s.delete( s.get( TShirt.class, ts.getId() ) );
+		tx.commit();
+
+		//Need to sleep for the message consumption
+		Thread.sleep( 500 );
+
+		assertEquals( 1, JGroupsReceiver.queues );
+		assertEquals( 1, JGroupsReceiver.works );
+		s.close();
+	}
+
+	private void prepareJGroupsChannel() throws Exception {
+		channel = new JChannel( XMLHelper.elementFromString( prepareXmlJGroupsConfiguration() ) );
+		channel.connect( CHANNEL_NAME );
+		channel.setReceiver( new JGroupsReceiver() );
+	}
+
+	protected void setUp() throws Exception {
+		super.setUp();
+		prepareJGroupsChannel();
+	}
+
+	protected void tearDown() throws Exception {
+		channel.close();
+		super.tearDown();
+	}
+
+	protected Class[] getMappings() {
+		return new Class[] {
+				TShirt.class
+		};
+	}
+
+	protected void configure(Configuration cfg) {
+		super.configure( cfg );
+		cfg.setProperty( Environment.WORKER_BACKEND, "jgroupsSlave" );
+		cfg.setProperty( JGroupsBackendQueueProcessorFactory.CONFIGURATION_XML, prepareXmlJGroupsConfiguration() );
+	}
+
+	private String prepareXmlJGroupsConfiguration() {
+		return "<config>" +
+				"<UDP" +
+				"     mcast_addr=\"${jgroups.udp.mcast_addr:228.10.10.10}\"" +
+				"     mcast_port=\"${jgroups.udp.mcast_port:45588}\"" +
+				"     tos=\"8\"" +
+				"     ucast_recv_buf_size=\"20000000\"" +
+				"     ucast_send_buf_size=\"640000\"" +
+				"     mcast_recv_buf_size=\"25000000\"" +
+				"     mcast_send_buf_size=\"640000\"" +
+				"     loopback=\"false\"\n" +
+				"     discard_incompatible_packets=\"true\"" +
+				"     max_bundle_size=\"64000\"" +
+				"     max_bundle_timeout=\"30\"" +
+				"     use_incoming_packet_handler=\"true\"" +
+				"     ip_ttl=\"${jgroups.udp.ip_ttl:2}\"" +
+				"     enable_bundling=\"true\"" +
+				"     enable_diagnostics=\"true\"" +
+				"     use_concurrent_stack=\"true\"" +
+				"     thread_naming_pattern=\"pl\"" +
+				"     thread_pool.enabled=\"true\"" +
+				"     thread_pool.min_threads=\"1\"" +
+				"     thread_pool.max_threads=\"25\"" +
+				"     thread_pool.keep_alive_time=\"5000\"" +
+				"     thread_pool.queue_enabled=\"false\"" +
+				"     thread_pool.queue_max_size=\"100\"" +
+				"     thread_pool.rejection_policy=\"Run\"" +
+				"     oob_thread_pool.enabled=\"true\"" +
+				"     oob_thread_pool.min_threads=\"1\"" +
+				"     oob_thread_pool.max_threads=\"8\"" +
+				"     oob_thread_pool.keep_alive_time=\"5000\"" +
+				"     oob_thread_pool.queue_enabled=\"false\"" +
+				"     oob_thread_pool.queue_max_size=\"100\"" +
+				"     oob_thread_pool.rejection_policy=\"Run\"/>" +
+				"<PING timeout=\"2000\" num_initial_members=\"3\"/>" +
+				"<MERGE2 max_interval=\"30000\" min_interval=\"10000\"/>" +
+				"<FD_SOCK/>" +
+				"<FD timeout=\"10000\" max_tries=\"5\" shun=\"true\"/>" +
+				"<VERIFY_SUSPECT timeout=\"1500\"/>" +
+				"<pbcast.NAKACK " +
+				"            use_mcast_xmit=\"false\" gc_lag=\"0\"" +
+				"            retransmit_timeout=\"300,600,1200,2400,4800\"" +
+				"            discard_delivered_msgs=\"false\"/>" +
+				"<UNICAST timeout=\"300,600,1200,2400,3600\"/>" +
+				"<pbcast.STABLE stability_delay=\"1000\" desired_avg_gossip=\"50000\"" +
+				"            max_bytes=\"400000\"/>   " +
+				"<pbcast.GMS print_local_addr=\"true\" join_timeout=\"3000\"" +
+				"            shun=\"false\"" +
+				"            view_bundling=\"true\"/>" +
+				"<FC max_credits=\"20000000\" min_threshold=\"0.10\"/>" +
+				"<FRAG2 frag_size=\"60000\"/>" +
+				"<pbcast.STREAMING_STATE_TRANSFER />" +
+				"<pbcast.FLUSH timeout=\"0\"/>" +
+				"</config>";
+	}
+}
Index: ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/slave/JGroupsReceiver.java
===================================================================
--- ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/slave/JGroupsReceiver.java	Mon Jun 22 23:41:13 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/slave/JGroupsReceiver.java	Mon Jun 22 23:41:13 CEST 2009
@@ -0,0 +1,39 @@
+package org.hibernate.search.test.jgroups.slave;
+
+import java.util.List;
+
+import org.jgroups.Message;
+import org.jgroups.ReceiverAdapter;
+
+import org.hibernate.search.backend.LuceneWork;
+
+/**
+ * @author Lukasz Moren
+ */
+
+public class JGroupsReceiver extends ReceiverAdapter {
+
+	public static int queues;
+	public static int works;
+
+	public static void reset() {
+		queues = 0;
+		works = 0;
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public void receive(Message message) {
+
+		List<LuceneWork> queue;
+		try {
+			queue = ( List<LuceneWork> ) message.getObject();
+		}
+
+		catch ( ClassCastException e ) {
+			return;
+		}
+		queues++;
+		works += queue.size();
+	}
+}
Index: ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/BatchedQueueingProcessor.java
===================================================================
--- ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/BatchedQueueingProcessor.java	(revision 16312)
+++ ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/BatchedQueueingProcessor.java	Mon Jun 22 23:34:55 CEST 2009
@@ -24,6 +24,8 @@
 import org.hibernate.search.backend.WorkType;
 import org.hibernate.search.backend.configuration.ConfigurationParseHelper;
 import org.hibernate.search.backend.impl.blackhole.BlackHoleBackendQueueProcessorFactory;
+import org.hibernate.search.backend.impl.jgroups.MasterJGroupsBackendQueueProcessorFactory;
+import org.hibernate.search.backend.impl.jgroups.SlaveJGroupsBackendQueueProcessorFactory;
 import org.hibernate.search.backend.impl.jms.JMSBackendQueueProcessorFactory;
 import org.hibernate.search.backend.impl.lucene.LuceneBackendQueueProcessorFactory;
 import org.hibernate.search.engine.DocumentBuilderIndexedEntity;
@@ -87,6 +89,12 @@
 		else if ( "blackhole".equalsIgnoreCase( backend ) ) {
 			backendQueueProcessorFactory = new BlackHoleBackendQueueProcessorFactory();
 		}
+		else if ( "jgroupsMaster".equals( backend ) ) {
+			backendQueueProcessorFactory = new MasterJGroupsBackendQueueProcessorFactory();
+		}
+		else if ( "jgroupsSlave".equals( backend ) ) {
+			backendQueueProcessorFactory = new SlaveJGroupsBackendQueueProcessorFactory();
+		}
 		else {
 			try {
 				Class processorFactoryClass = ReflectHelper.classForName( backend, BatchedQueueingProcessor.class );
Index: ../HibernateSearch/pom.xml
===================================================================
--- ../HibernateSearch/pom.xml	(revision 16784)
+++ ../HibernateSearch/pom.xml	Tue Jun 23 21:10:27 CEST 2009
@@ -132,7 +132,7 @@
             </plugin>
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-jar-plugin</artifactId>              
+                <artifactId>maven-jar-plugin</artifactId>
                 <configuration>
                     <archive>
                         <manifestEntries>
@@ -156,6 +156,24 @@
                             <name>build.dir</name>
                             <value>${basedir}/target</value>
                         </property>
+                        <!--
+                            Following is the default jgroups mcast address. If you find the testsuite runs very slowly,
+                            there may be problems with multicast on the interface JGroups uses by default on
+                            your machine. You can try to resolve setting 'jgroups.bind_addr' as a system-property
+                            to the jvm launching maven and setting the value to an interface where you know multicast works
+                        -->
+                        <property>
+                            <name>jgroups.bind_addr</name>
+                            <value>127.0.0.1</value>
+                        </property>
+                        <!-- There are problems with multicast and IPv6 on some OS/JDK combos, so we tell Java
+                             to use IPv4. If you have problems with multicast when running the tests you can
+                             try setting this to 'false', although typically that won't be helpful.
+                        -->
+                        <property>
+                            <name>java.net.preferIPv4Stack</name>
+                            <value>true</value>
+                        </property>
                     </systemProperties>
                     <excludes>
                         <exclude>**/*.java</exclude>
@@ -181,7 +199,7 @@
                     <masterTranslation>en-US</masterTranslation>
                     <translations>
                         <translation>zh-CN</translation>
-                    </translations>                         
+                    </translations>
                     <imageResource>
                         <directory>${basedir}/src/main/docbook/en-US/images</directory>
                     </imageResource>
@@ -658,6 +676,12 @@
                     <optional>true</optional>
                 </dependency>
                 <dependency>
+                    <groupId>jgroups</groupId>
+                    <artifactId>jgroups</artifactId>
+                    <version>2.6.7.GA</version>
+                    <optional>true</optional>
+                </dependency>
+                <dependency>
                     <groupId>javax.annotation</groupId>
                     <artifactId>jsr250-api</artifactId>
                     <version>1.0</version>
Index: ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/slave/TShirt.java
===================================================================
--- ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/slave/TShirt.java	Wed Jun 10 00:09:04 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/slave/TShirt.java	Wed Jun 10 00:09:04 CEST 2009
@@ -0,0 +1,50 @@
+
+package org.hibernate.search.test.jgroups.slave;
+
+import org.hibernate.search.annotations.DocumentId;
+import org.hibernate.search.annotations.Field;
+import org.hibernate.search.annotations.Index;
+import org.hibernate.search.annotations.Indexed;
+
+import javax.persistence.Entity;
+import javax.persistence.GeneratedValue;
+import javax.persistence.Id;
+
+/**
+ * @author Emmanuel Bernard
+ */
+@Entity
+@Indexed
+public class TShirt {
+	@Id
+	@GeneratedValue
+	@DocumentId
+	private int id;
+	@Field(index= Index.TOKENIZED)
+	private String logo;
+	private String size;
+
+	public int getId() {
+		return id;
+	}
+
+	public void setId(int id) {
+		this.id = id;
+	}
+
+	public String getLogo() {
+		return logo;
+	}
+
+	public void setLogo(String logo) {
+		this.logo = logo;
+	}
+
+	public String getSize() {
+		return size;
+	}
+
+	public void setSize(String size) {
+		this.size = size;
+	}
+}
\ No newline at end of file
Index: ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/master/JGroupsMasterTest.java
===================================================================
--- ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/master/JGroupsMasterTest.java	Tue Jun 23 21:03:43 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/master/JGroupsMasterTest.java	Tue Jun 23 21:03:43 CEST 2009
@@ -0,0 +1,170 @@
+package org.hibernate.search.test.jgroups.master;
+
+import java.io.Serializable;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.lucene.analysis.StopAnalyzer;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.queryParser.QueryParser;
+import org.apache.lucene.search.Query;
+import org.jgroups.JChannel;
+import org.jgroups.Message;
+
+import org.hibernate.HibernateException;
+import org.hibernate.Session;
+import org.hibernate.cfg.Configuration;
+import org.hibernate.search.Environment;
+import org.hibernate.search.FullTextSession;
+import org.hibernate.search.Search;
+import org.hibernate.search.backend.AddLuceneWork;
+import org.hibernate.search.backend.LuceneWork;
+import org.hibernate.search.backend.impl.jgroups.JGroupsBackendQueueProcessorFactory;
+import org.hibernate.search.engine.DocumentBuilder;
+import org.hibernate.search.test.SearchTestCase;
+import org.hibernate.search.test.jms.master.TShirt;
+
+/**
+ * Tests that the Master node in a JGroups cluster can properly process messages received from channel.
+ * <p/>
+ * In case of running test outside Hibernate Search Maven configuration set following VM configuration:
+ * <br><br>
+ * <code>
+ * -Djava.net.preferIPv4Stack=true -Djgroups.bind_addr=127.0.0.1
+ * </code>
+ *
+ * @author Lukasz Moren
+ */
+public class JGroupsMasterTest extends SearchTestCase {
+
+	/**
+	 * Name of the JGroups channel used in test
+	 */
+	public static final String CHANNEL_NAME = "jgroups_test_channel";
+
+	private JChannel channel;
+
+	public void testMessageSending() throws Exception {
+
+		TShirt shirt = createObjectWithSQL();
+		List<LuceneWork> queue = createDocumentAndWorkQueue( shirt );
+
+		sendMessage( queue );
+
+		Thread.sleep( 3000 );
+
+		FullTextSession ftSess = Search.getFullTextSession( openSession() );
+		ftSess.getTransaction().begin();
+		QueryParser parser = new QueryParser( "id", new StopAnalyzer() );
+		Query luceneQuery = parser.parse( "logo:jboss" );
+		org.hibernate.Query query = ftSess.createFullTextQuery( luceneQuery );
+		List result = query.list();
+		assertEquals( 1, result.size() );
+		ftSess.delete( result.get( 0 ) );
+		ftSess.getTransaction().commit();
+		ftSess.close();
+	}
+
+	private void prepareJGroupsChannel() throws Exception {
+		channel = new JChannel( prepareJGroupsCongigurationString() );
+		channel.connect( CHANNEL_NAME );
+	}
+
+	private void sendMessage(List<LuceneWork> queue) throws Exception {
+		//send message to all listeners
+		Message message = new Message( null, null, ( Serializable ) queue );
+		channel.send( message );
+	}
+
+	/**
+	 * Manually create the work queue. This lists gets send by the Slaves to the Master for indexing.
+	 *
+	 * @param shirt The shirt to index
+	 *
+	 * @return A manually create <code>LuceneWork</code> list.
+	 */
+	private List<LuceneWork> createDocumentAndWorkQueue(TShirt shirt) {
+		Document doc = new Document();
+		Field field = new Field(
+				DocumentBuilder.CLASS_FIELDNAME, shirt.getClass().getName(), Field.Store.YES, Field.Index.NOT_ANALYZED
+		);
+		doc.add( field );
+		field = new Field( "id", "1", Field.Store.YES, Field.Index.NOT_ANALYZED );
+		doc.add( field );
+		field = new Field( "logo", shirt.getLogo(), Field.Store.NO, Field.Index.ANALYZED );
+		doc.add( field );
+		LuceneWork luceneWork = new AddLuceneWork(
+				shirt.getId(), String.valueOf( shirt.getId() ), shirt.getClass(), doc
+		);
+		List<LuceneWork> queue = new ArrayList<LuceneWork>();
+		queue.add( luceneWork );
+		return queue;
+	}
+
+	/**
+	 * Create a test object without trigggering indexing. Use SQL directly.
+	 *
+	 * @return a <code>TShirt</code> test object.
+	 *
+	 * @throws java.sql.SQLException in case the inset fails.
+	 */
+	@SuppressWarnings({ "deprecation" })
+	private TShirt createObjectWithSQL() throws SQLException {
+		Session s = openSession();
+		s.getTransaction().begin();
+		Statement statement = s.connection().createStatement();
+		statement.executeUpdate(
+				"insert into TShirt_Master(id, logo, size) values( '1', 'JBoss balls', 'large')"
+		);
+		statement.close();
+		TShirt ts = ( TShirt ) s.get( TShirt.class, 1 );
+		s.getTransaction().commit();
+		s.close();
+		return ts;
+	}
+
+	public static Session getSession() throws HibernateException {
+		return sessions.openSession();
+	}
+
+	protected void setUp() throws Exception {
+		prepareJGroupsChannel();
+		super.setUp();
+	}
+
+	protected void tearDown() throws Exception {
+		channel.close();
+		super.tearDown();
+	}
+
+	protected void configure(Configuration cfg) {
+		super.configure( cfg );
+		// JGroups configuration for master node
+		cfg.setProperty( Environment.WORKER_BACKEND, "jgroupsMaster" );
+		cfg.setProperty( JGroupsBackendQueueProcessorFactory.JG_CLUSTER_NAME, CHANNEL_NAME );
+		cfg.setProperty(
+				JGroupsBackendQueueProcessorFactory.CONFIGURATION_STRING, prepareJGroupsCongigurationString()
+		);
+	}
+
+	private String prepareJGroupsCongigurationString() {
+		return "UDP(mcast_addr=228.1.2.3;mcast_port=45566;ip_ttl=32):" +
+				"PING(timeout=3000;num_initial_members=6):" +
+				"FD(timeout=5000):" +
+				"VERIFY_SUSPECT(timeout=1500):" +
+				"pbcast.NAKACK(gc_lag=10;retransmit_timeout=3000):" +
+				"UNICAST(timeout=5000):" +
+				"FRAG:" +
+				"pbcast.GMS(join_timeout=3000;" +
+				"shun=false;print_local_addr=true)";
+	}
+
+	protected Class[] getMappings() {
+		return new Class[] {
+				TShirt.class
+		};
+	}
+}
\ No newline at end of file
Index: ../HibernateSearch/src/test/java/org/hibernate/search/test/SearchTestCase.java
===================================================================
--- ../HibernateSearch/src/test/java/org/hibernate/search/test/SearchTestCase.java	(revision 16579)
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/SearchTestCase.java	Sat Jun 20 19:22:58 CEST 2009
@@ -153,4 +153,8 @@
 	protected String[] getAnnotatedPackages() {
 		return new String[] { };
 	}
+
+	protected static File getIndexDir() {
+		return indexDir;
-}
+	}
+}
Index: ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/SlaveJGroupsBackendQueueProcessorFactory.java
===================================================================
--- ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/SlaveJGroupsBackendQueueProcessorFactory.java	Mon Jun 22 23:44:19 CEST 2009
+++ ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/SlaveJGroupsBackendQueueProcessorFactory.java	Mon Jun 22 23:44:19 CEST 2009
@@ -0,0 +1,15 @@
+package org.hibernate.search.backend.impl.jgroups;
+
+import java.util.List;
+
+import org.hibernate.search.backend.LuceneWork;
+
+/**
+ * @author Lukasz Moren
+ */
+public class SlaveJGroupsBackendQueueProcessorFactory extends JGroupsBackendQueueProcessorFactory {
+
+	public Runnable getProcessor(List<LuceneWork> queue) {
+		return new JGroupsBackendQueueProcessor( queue, this );
+	}
+}
Index: ../HibernateSearch/src/main/java/org/hibernate/search/util/XMLHelper.java
===================================================================
--- ../HibernateSearch/src/main/java/org/hibernate/search/util/XMLHelper.java	Mon Jun 22 22:08:16 CEST 2009
+++ ../HibernateSearch/src/main/java/org/hibernate/search/util/XMLHelper.java	Mon Jun 22 22:08:16 CEST 2009
@@ -0,0 +1,35 @@
+package org.hibernate.search.util;
+
+import java.io.ByteArrayInputStream;
+import javax.xml.parsers.DocumentBuilder;
+import javax.xml.parsers.DocumentBuilderFactory;
+
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+
+/**
+ * A utility class to help with xml parsing
+ *
+ * @author Lukasz Moren
+ */
+public class XMLHelper {
+
+
+	/**
+	 * Converts a String representing an XML snippet into an {@link org.w3c.dom.Element}.
+	 *
+	 * @param xml snippet as a string
+	 *
+	 * @return a DOM Element
+	 *
+	 * @throws Exception if unable to parse the String or if it doesn't contain valid XML.
+	 */
+	public static Element elementFromString(String xml) throws Exception {
+		ByteArrayInputStream bais = new ByteArrayInputStream( xml.getBytes( "UTF-8" ) );
+		DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
+		DocumentBuilder builder = factory.newDocumentBuilder();
+		Document document = builder.parse( bais );
+		bais.close();
+		return document.getDocumentElement();
+	}
+}
Index: ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsMasterMessageListener.java
===================================================================
--- ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsMasterMessageListener.java	Mon Jun 22 23:38:42 CEST 2009
+++ ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsMasterMessageListener.java	Mon Jun 22 23:38:42 CEST 2009
@@ -0,0 +1,87 @@
+package org.hibernate.search.backend.impl.jgroups;
+
+import java.util.List;
+
+import org.jgroups.Address;
+import org.jgroups.Message;
+import org.jgroups.Receiver;
+import org.jgroups.View;
+import org.slf4j.Logger;
+
+import org.hibernate.search.backend.LuceneWork;
+import org.hibernate.search.engine.SearchFactoryImplementor;
+import org.hibernate.search.util.LoggerFactory;
+
+/**
+ * Listen for messages from slave nodes and apply them into <code>LuceneBackendQueueProcessor</code>
+ *
+ * @author Lukasz Moren
+ * @see org.hibernate.search.backend.impl.lucene.LuceneBackendQueueProcessorFactory
+ * @see org.hibernate.search.backend.impl.lucene.LuceneBackendQueueProcessor
+ * @see org.jgroups.Receiver
+ */
+public class JGroupsMasterMessageListener implements Receiver {
+
+	private static final Logger log = LoggerFactory.make();
+
+	private SearchFactoryImplementor searchFactory;
+
+	public JGroupsMasterMessageListener(SearchFactoryImplementor searchFactory) {
+		this.searchFactory = searchFactory;
+	}
+
+	@SuppressWarnings("unchecked")
+	public void receive(Message message) {
+		List<LuceneWork> queue;
+		try {
+			queue = ( List<LuceneWork> ) message.getObject();
+		}
+		catch ( ClassCastException e ) {
+			log.error( "Illegal object retrieved from message.", e );
+			return;
+		}
+
+		if ( queue != null && !queue.isEmpty() ) {
+			log.debug(
+					"There are {} Lucene docs received from slave node {} to be processed by master",
+					queue.size(),
+					message.getSrc()
+			);
+			Runnable worker = getWorker( queue );
+			worker.run();
+		}
+		else {
+			log.warn( "Received null or empty Lucene works list in message." );
+		}
+	}
+
+	private Runnable getWorker(List<LuceneWork> queue) {
+		Runnable processor;
+		processor = searchFactory.getBackendQueueProcessorFactory().getProcessor( queue );
+		return processor;
+	}
+
+	// ------------------------------------------------------------------------------------------------------------------
+	// Implementations of JGroups interfaces
+	// ------------------------------------------------------------------------------------------------------------------
+
+	public byte[] getState() {
+		return null;
+	}
+
+	public void setState(byte[] state) {
+		//no-op
+	}
+
+	public void viewAccepted(View view) {
+		log.info( "Received new cluster view: {}", view );
+	}
+
+	public void suspect(Address suspected_mbr) {
+		//no-op
+	}
+
+	public void block() {
+		//no-op
+	}
+}
