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	Sun Jun 07 19:59:31 CEST 2009
+++ ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsBackendQueueProcessor.java	Sun Jun 07 19:59:31 CEST 2009
@@ -0,0 +1,58 @@
+package org.hibernate.search.backend.impl.jgroups;
+
+import org.hibernate.search.SearchException;
+import org.hibernate.search.backend.LuceneWork;
+import org.hibernate.search.backend.OptimizeLuceneWork;
+import org.jgroups.ChannelClosedException;
+import org.jgroups.ChannelNotConnectedException;
+import org.jgroups.Message;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author Lukasz Moren
+ */
+public class JGroupsBackendQueueProcessor implements Runnable {
+
+	private JGroupsBackendQueueProcessorFactory factory;
+	private 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 );
+		for ( LuceneWork work : queue ) {
+			if ( work instanceof OptimizeLuceneWork ) {
+				//we don't want optimization to be propagated
+				filteredQueue.remove( work );
+			}
+		}
+		if ( filteredQueue.size() == 0 ) {
+			return;
+		}
+
+		try {
+			
+			//send message with lucene works to all receivers in cluster
+			Message message = new Message( null, null, ( Serializable ) filteredQueue );
+			factory.getChannel().send( message );
+
+		}
+		catch ( ChannelNotConnectedException e ) {
+			throw new SearchException(
+					"Unable to send Lucene work. Channel is not connected to: "
+							+ factory.getChannel().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/JGroupsBackendQueueProcessorFactory.java
===================================================================
--- ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsBackendQueueProcessorFactory.java	Wed Jun 10 00:09:04 CEST 2009
+++ ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsBackendQueueProcessorFactory.java	Wed Jun 10 00:09:04 CEST 2009
@@ -0,0 +1,99 @@
+package org.hibernate.search.backend.impl.jgroups;
+
+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.backend.impl.lucene.LuceneBackendQueueProcessorFactory;
+import org.hibernate.search.engine.SearchFactoryImplementor;
+import org.hibernate.util.ReflectHelper;
+import org.jgroups.Channel;
+import org.jgroups.ChannelException;
+import org.jgroups.JChannel;
+import org.jgroups.Receiver;
+
+import java.util.List;
+import java.util.Properties;
+
+/**
+ * @author Lukasz Moren
+ */
+public class JGroupsBackendQueueProcessorFactory implements BackendQueueProcessorFactory {
+
+	private String listenerClassName;
+	private String clusterName;
+	private Properties cfg;
+	private SearchFactoryImplementor searchFactory;
+	private Receiver receiver;
+	private JChannel channel = null;
+	private LuceneBackendQueueProcessorFactory luceneBackendQueueProcessorFactory;
+
+	public static final String JG_CHANNEL_NAME = Environment.WORKER_PREFIX + "jgroups.channel_name";
+	public static final String JGROUPS_LISTENER_CLASS = Environment.WORKER_PREFIX + "jgroups.listener_class";
+
+	public void initialize(Properties props, SearchFactoryImplementor searchFactory) {
+		this.cfg = props;
+		this.searchFactory = searchFactory;
+		this.clusterName = props.getProperty( JG_CHANNEL_NAME );
+		this.listenerClassName = props.getProperty( JGROUPS_LISTENER_CLASS );
+
+		prepareJGroupsChannel();
+		registerChannelReceiver();
+	}
+
+	public Runnable getProcessor(List<LuceneWork> queue) {
+		//if it is a master node return lucene processor to not send works again
+		if ( receiver == null ) {
+			return new JGroupsBackendQueueProcessor( queue, this );
+		}
+		return luceneBackendQueueProcessorFactory.getProcessor( queue );
+	}
+
+	private void prepareJGroupsChannel() {
+		try {
+			channel = new JChannel();
+			channel.setOpt( Channel.AUTO_RECONNECT, Boolean.TRUE);
+			channel.connect( clusterName );
+		}
+		catch ( ChannelException e ) {
+			throw new SearchException( "Unable to connect to: " + clusterName + "JGroups channel" );
+		}
+	}
+
+	private void registerChannelReceiver() {
+		try {
+			if ( listenerClassName != null ) {
+
+				Class clazz = ReflectHelper.classForName( listenerClassName );
+				receiver = ( Receiver ) clazz.newInstance();
+				channel.setReceiver( receiver );
+
+				//in case it is master node init Lucene procssor factory
+				initLuceneBackendQueueProcessorFactory();
+			}
+		}
+		catch ( ClassNotFoundException e ) {
+			throw new SearchException( "Unable to find JGroups receiver class: " + listenerClassName, e );
+		}
+		catch ( IllegalAccessException e ) {
+			throw new SearchException( "Unable to instanciate receiver class: " + listenerClassName, e );
+		}
+		catch ( InstantiationException e ) {
+			throw new SearchException( "Unable to instanciate receiver class: " + listenerClassName, e );
+		}
+
+	}
+
+	private void initLuceneBackendQueueProcessorFactory() {
+		luceneBackendQueueProcessorFactory = new LuceneBackendQueueProcessorFactory();
+		luceneBackendQueueProcessorFactory.initialize( cfg, searchFactory );
+	}
+
+	public void close() {
+		channel.disconnect();
+	}
+
+	public JChannel getChannel() {
+		return channel;
+	}
+}
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	Tue Jun 09 23:56:08 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/common/MultipleSessionsSearchTestCase.java	Tue Jun 09 23:56:08 CEST 2009
@@ -0,0 +1,141 @@
+package org.hibernate.search.test.jgroups.common;
+
+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;
+import org.slf4j.Logger;
+
+import java.io.InputStream;
+
+/**
+ * @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", indexDir.getAbsolutePath() + masterCopy );
+		cfg.setProperty( "hibernate.search.default.indexBase", indexDir.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", indexDir.getAbsolutePath() + masterCopy );
+		cfg.setProperty( "hibernate.search.default.indexBase", indexDir.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 ( indexDir.exists() ) {
+			FileHelper.delete( indexDir );
+		}
+		super.setUp();
+		buildCommonSessionFactory( getCommonMappings(), getCommonAnnotatedPackages(), getCommonXmlFiles() );
+	}
+
+	@Override
+	protected void tearDown() throws Exception {
+		super.tearDown();
+		log.info( "Deleting test directory {} ", indexDir.getAbsolutePath() );
+		FileHelper.delete( indexDir );
+		slaveSessionFactory = null;
+	}
+
+	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	Tue Jun 09 22:55:12 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/common/JGroupsCommonTest.java	Tue Jun 09 22:55:12 CEST 2009
@@ -0,0 +1,128 @@
+package org.hibernate.search.test.jgroups.common;
+
+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;
+
+import java.util.List;
+
+/**
+ * @author Lukasz Moren
+ */
+
+public class JGroupsCommonTest extends MultipleSessionsSearchTestCase {
+
+	public static final String CHANNEL_NAME = "jgroups_test_channel";
+
+	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, "jgroups" );
+		cfg.setProperty(
+				JGroupsBackendQueueProcessorFactory.JGROUPS_LISTENER_CLASS, JGroupsMessageReceiverImpl.class.getName()
+		);
+		cfg.setProperty( JGroupsBackendQueueProcessorFactory.JG_CHANNEL_NAME, CHANNEL_NAME );
+
+
+	}
+
+	@Override
+	protected void commonConfigure(Configuration cfg) {
+		//slave jgroups configuration
+		super.commonConfigure( cfg );
+		cfg.setProperty( Environment.WORKER_BACKEND, "jgroups" );
+		cfg.setProperty( JGroupsBackendQueueProcessorFactory.JG_CHANNEL_NAME, CHANNEL_NAME );
+
+
+	}
+
+	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/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsAbstractMessageReceiver.java
===================================================================
--- ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsAbstractMessageReceiver.java	Tue Jun 09 23:46:57 CEST 2009
+++ ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/jgroups/JGroupsAbstractMessageReceiver.java	Tue Jun 09 23:46:57 CEST 2009
@@ -0,0 +1,72 @@
+package org.hibernate.search.backend.impl.jgroups;
+
+import org.hibernate.Session;
+import org.hibernate.search.FullTextSession;
+import org.hibernate.search.Search;
+import org.hibernate.search.backend.LuceneWork;
+import org.hibernate.search.engine.SearchFactoryImplementor;
+import org.hibernate.search.util.LoggerFactory;
+import org.jgroups.Address;
+import org.jgroups.Message;
+import org.jgroups.Receiver;
+import org.jgroups.View;
+import org.slf4j.Logger;
+
+import java.util.List;
+
+/**
+ * @author Lukasz Moren
+ */
+public abstract class JGroupsAbstractMessageReceiver implements Receiver {
+
+	private static final Logger log = LoggerFactory.make();
+
+	@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;
+		}
+		Runnable worker = getWorker( queue );
+		worker.run();
+	}
+
+	public abstract Session getSession();
+
+	protected abstract void cleanSessionIfNeeded(Session session);
+
+	private Runnable getWorker(List<LuceneWork> queue) {
+		Session session = getSession();
+		Runnable processor = null;
+		try {
+			FullTextSession fullTextSession = Search.getFullTextSession( session );
+			SearchFactoryImplementor searchFactory = ( SearchFactoryImplementor ) fullTextSession.getSearchFactory();
+			processor = searchFactory.getBackendQueueProcessorFactory().getProcessor( queue );
+		}
+		finally {
+			cleanSessionIfNeeded( session );
+		}
+		return processor;
+	}
+
+	public byte[] getState() {
+		return null;
+	}
+
+	public void setState(byte[] state) {
+	}
+
+	public void viewAccepted(View view) {
+		log.info( "Channel changed " + view + " nodes are connectd to the cluster " );
+	}
+
+	public void suspect(Address suspected_mbr) {
+	}
+
+	public void block() {
+	}
+}
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 09 23:50:07 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/slave/JGroupsSlaveTest.java	Tue Jun 09 23:50:07 CEST 2009
@@ -0,0 +1,103 @@
+package org.hibernate.search.test.jgroups.slave;
+
+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.jgroups.Channel;
+import org.jgroups.JChannel;
+
+/**
+ * Tests that the Slave node in a JGroups cluster can properly send messages to the channel.
+ *
+ * @author Lukasz Moren
+ */
+
+public class JGroupsSlaveTest extends SearchTestCase {
+
+	public static final String CHANNEL_NAME = "jgroups_test_channel";
+
+	private JChannel 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() ) );
+		s.delete( s.get( TShirt.class, ts2.getId() ) );
+		tx.commit();
+
+		//Need to sleep for the message consumption
+		Thread.sleep( 500 );
+
+		assertEquals( 1, JGroupsReceiver.queues );
+		assertEquals( 2, JGroupsReceiver.works );
+		s.close();
+	}
+
+	private void prepareJGroupsChannel() throws Exception {
+		channel = new JChannel();
+		channel.setOpt( Channel.AUTO_RECONNECT, Boolean.TRUE );
+		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 void configure(Configuration cfg) {
+		super.configure( cfg );
+		cfg.setProperty( Environment.WORKER_BACKEND, "jgroups" );
+		cfg.setProperty( JGroupsBackendQueueProcessorFactory.JG_CHANNEL_NAME, CHANNEL_NAME );
+	}
+
+	protected Class[] getMappings() {
+		return new Class[] {
+				TShirt.class
+		};
+	}
+}
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	Sun Jun 07 21:59:47 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/slave/JGroupsReceiver.java	Sun Jun 07 21:59:47 CEST 2009
@@ -0,0 +1,38 @@
+package org.hibernate.search.test.jgroups.slave;
+
+import org.hibernate.search.backend.LuceneWork;
+import org.jgroups.Message;
+import org.jgroups.ReceiverAdapter;
+
+import java.util.List;
+
+/**
+ * @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 16668)
+++ ../HibernateSearch/src/main/java/org/hibernate/search/backend/impl/BatchedQueueingProcessor.java	Wed Jun 10 00:46:24 CEST 2009
@@ -24,6 +24,7 @@
 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.JGroupsBackendQueueProcessorFactory;
 import org.hibernate.search.backend.impl.jms.JMSBackendQueueProcessorFactory;
 import org.hibernate.search.backend.impl.lucene.LuceneBackendQueueProcessorFactory;
 import org.hibernate.search.engine.DocumentBuilderIndexedEntity;
@@ -87,6 +88,9 @@
 		else if ( "blackhole".equalsIgnoreCase( backend ) ) {
 			backendQueueProcessorFactory = new BlackHoleBackendQueueProcessorFactory();
 		}
+        else if("jgroups".equals(backend)) {
+            backendQueueProcessorFactory = new JGroupsBackendQueueProcessorFactory();
+        }
 		else {
 			try {
 				Class processorFactoryClass = ReflectHelper.classForName( backend, BatchedQueueingProcessor.class );
Index: ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/master/JGroupsMessageReceiverImpl.java
===================================================================
--- ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/master/JGroupsMessageReceiverImpl.java	Wed Jun 10 00:09:04 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/master/JGroupsMessageReceiverImpl.java	Wed Jun 10 00:09:04 CEST 2009
@@ -0,0 +1,22 @@
+
+package org.hibernate.search.test.jgroups.master;
+
+import org.hibernate.Session;
+import org.hibernate.search.backend.impl.jgroups.JGroupsAbstractMessageReceiver;
+
+/**
+ * @author Lukasz Moren
+ */
+
+public class JGroupsMessageReceiverImpl extends JGroupsAbstractMessageReceiver {
+
+	public Session getSession() {
+		return JGroupsMasterTest.getSession();
+    }
+
+    protected void cleanSessionIfNeeded(Session session) {
+        session.close();
+    }
+
+
+}
Index: ../HibernateSearch/pom.xml
===================================================================
--- ../HibernateSearch/pom.xml	(revision 16668)
+++ ../HibernateSearch/pom.xml	Tue Jun 09 23:59:12 CEST 2009
@@ -84,6 +84,11 @@
             <artifactId>jta</artifactId>
             <version>1.1</version>
         </dependency>
+        <dependency>
+            <groupId>jgroups</groupId>
+            <artifactId>jgroups</artifactId>
+            <version>2.6.7.GA</version>
+        </dependency>
 
         <!-- =============================== -->
         <!-- Testing Dependencies            -->
Index: ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/common/JGroupsMessageReceiverImpl.java
===================================================================
--- ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/common/JGroupsMessageReceiverImpl.java	Wed Jun 10 00:09:04 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/common/JGroupsMessageReceiverImpl.java	Wed Jun 10 00:09:04 CEST 2009
@@ -0,0 +1,22 @@
+
+package org.hibernate.search.test.jgroups.common;
+
+import org.hibernate.Session;
+import org.hibernate.search.backend.impl.jgroups.JGroupsAbstractMessageReceiver;
+
+/**
+ * @author Lukasz Moren
+ */
+
+public class JGroupsMessageReceiverImpl extends JGroupsAbstractMessageReceiver {
+
+	public Session getSession() {
+		return JGroupsCommonTest.getSession();
+    }
+
+    protected void cleanSessionIfNeeded(Session session) {
+        session.close();
+    }
+
+
+}
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	Wed Jun 10 00:10:11 CEST 2009
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/jgroups/master/JGroupsMasterTest.java	Wed Jun 10 00:10:11 CEST 2009
@@ -0,0 +1,153 @@
+package org.hibernate.search.test.jgroups.master;
+
+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.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;
+import org.jgroups.Channel;
+import org.jgroups.JChannel;
+import org.jgroups.Message;
+
+import java.io.Serializable;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Tests that the Master node in a JGroups cluster can properly process messages received from channel.
+ *
+ * @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( 2000 );
+
+		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();
+		channel.setOpt( Channel.AUTO_RECONNECT, Boolean.TRUE );
+		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, "jgroups" );
+		cfg.setProperty(
+				JGroupsBackendQueueProcessorFactory.JGROUPS_LISTENER_CLASS, JGroupsMessageReceiverImpl.class.getName()
+		);
+		cfg.setProperty( JGroupsBackendQueueProcessorFactory.JG_CHANNEL_NAME, CHANNEL_NAME );
+	}
+
+	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 16668)
+++ ../HibernateSearch/src/test/java/org/hibernate/search/test/SearchTestCase.java	Wed Jun 10 00:45:01 CEST 2009
@@ -34,7 +34,7 @@
 
 	private static final Logger log = org.hibernate.search.util.LoggerFactory.make();
 
-	private static File indexDir;
+	protected static File indexDir;
 	
 	private SearchFactory searchFactory;
 
