more refinements for Group

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@684047 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/group/DefaultMapChangedListener.java

@@ -22,12 +22,12 @@ package org.apache.activemq.group;
  */
 public class DefaultMapChangedListener implements MapChangedListener{
 
-    public void mapInsert(Member owner, Object key, Object value) {        
+    public void mapInsert(Member owner, Object key, Object value) {   
     }
 
-    public void mapRemove(Member owner, Object key, Object value,boolean expired) {        
+    public void mapRemove(Member owner, Object key, Object value,boolean expired) {  
     }
 
-    public void mapUpdate(Member owner, Object Key, Object oldValue,Object newValue) {        
+    public void mapUpdate(Member owner, Object key, Object oldValue,Object newValue) {
     }
 }

activemq-core/src/main/java/org/apache/activemq/group/EntryKey.java

@@ -28,9 +28,11 @@ import java.io.ObjectOutput;
 class EntryKey<K> implements Externalizable {
     private Member owner;
     private K key;
-    private boolean share;
+    private boolean locked;
     private boolean removeOnExit;
+    private boolean releaseLockOnExit;
     private long expiration;
+    private long lockExpiration;
 
     /**
      * Default constructor - for serialization
@@ -53,6 +55,10 @@ class EntryKey<K> implements Externalizable {
     public Member getOwner() {
         return this.owner;
     }
+    
+    public void setOwner(Member member) {
+        this.owner=member;
+    }
 
     /**
      * @return the key
@@ -64,15 +70,15 @@ class EntryKey<K> implements Externalizable {
     /**
      * @return the share
      */
-    public boolean isShare() {
+    public boolean isLocked() {
-        return this.share;
+        return this.locked;
     }
 
     /**
      * @param share the share to set
      */
-    public void setShare(boolean share) {
+    public void setLocked(boolean locked) {
-        this.share = share;
+        this.locked = locked;
     }
 
     /**
@@ -104,6 +110,34 @@ class EntryKey<K> implements Externalizable {
         this.expiration = expiration;
     }
     
+    /**
+     * @return the lockExpiration
+     */
+    public long getLockExpiration() {
+        return lockExpiration;
+    }
+
+    /**
+     * @param lockExpiration the lockExpiration to set
+     */
+    public void setLockExpiration(long lockExpiration) {
+        this.lockExpiration = lockExpiration;
+    }
+
+    /**
+     * @return the releaseLockOnExit
+     */
+    public boolean isReleaseLockOnExit() {
+        return releaseLockOnExit;
+    }
+
+    /**
+     * @param releaseLockOnExit the releaseLockOnExit to set
+     */
+    public void setReleaseLockOnExit(boolean releaseLockOnExit) {
+        this.releaseLockOnExit = releaseLockOnExit;
+    }
+    
     void setTimeToLive(long ttl) {
         if (ttl > 0 ) {
             this.expiration=ttl+System.currentTimeMillis();
@@ -112,6 +146,14 @@ class EntryKey<K> implements Externalizable {
         }
     }
     
+    void setLockTimeToLive(long ttl) {
+        if(ttl > 0) {
+            this.lockExpiration=ttl+System.currentTimeMillis();
+        }else {
+            this.lockExpiration=0l;
+        }
+    }
+    
     boolean isExpired() {
         return isExpired(System.currentTimeMillis());
     }
@@ -120,6 +162,14 @@ class EntryKey<K> implements Externalizable {
         return this.expiration > 0 && this.expiration < currentTime;
     }
     
+    boolean isLockExpired() {
+        return isLockExpired(System.currentTimeMillis());
+    }
+    
+    boolean isLockExpired(long currentTime) {
+        return this.lockExpiration > 0 && this.lockExpiration < currentTime;
+    }
+    
    
 
     public boolean equals(Object obj) {
@@ -134,21 +184,27 @@ class EntryKey<K> implements Externalizable {
     public void writeExternal(ObjectOutput out) throws IOException {
         out.writeObject(this.owner);
         out.writeObject(this.key);
-        out.writeBoolean(isShare());
+        out.writeBoolean(isLocked());
         out.writeBoolean(isRemoveOnExit());
+        out.writeBoolean(isReleaseLockOnExit());
         out.writeLong(getExpiration());
+        out.writeLong(getLockExpiration());
     }
 
     public void readExternal(ObjectInput in) throws IOException,
             ClassNotFoundException {
         this.owner = (Member) in.readObject();
         this.key = (K) in.readObject();
-        this.share = in.readBoolean();
+        this.locked = in.readBoolean();
         this.removeOnExit=in.readBoolean();
+        this.releaseLockOnExit=in.readBoolean();
         this.expiration=in.readLong();
+        this.lockExpiration=in.readLong();
     }
     
     public String toString() {
         return "key:"+this.key;
     }
+
+    
 }

activemq-core/src/main/java/org/apache/activemq/group/EntryMessage.java

@@ -29,9 +29,12 @@ public class EntryMessage implements Externalizable{
     static enum MessageType{INSERT,DELETE,SYNC};
     private EntryKey key;
     private Object value;
+    private Object oldValue;
     private MessageType type;
     private boolean mapUpdate;
     private boolean expired;
+    private boolean lockExpired;
+    private boolean lockUpdate;
     
     /**
      * @return the owner
@@ -58,6 +61,19 @@ public class EntryMessage implements Externalizable{
         this.value = value;
     }
     
+    /**
+     * @return the oldValue
+     */
+    public Object getOldValue() {
+        return this.oldValue;
+    }
+    /**
+     * @param oldValue the oldValue to set
+     */
+    public void setOldValue(Object oldValue) {
+        this.oldValue = oldValue;
+    }
+    
     /**
      * @return the type
      */
@@ -97,6 +113,32 @@ public class EntryMessage implements Externalizable{
         this.expired = expired;
     }
     
+    /**
+     * @return the lockExpired
+     */
+    public boolean isLockExpired() {
+        return lockExpired;
+    }
+    /**
+     * @param lockExpired the lockExpired to set
+     */
+    public void setLockExpired(boolean lockExpired) {
+        this.lockExpired = lockExpired;
+    }
+    
+    /**
+     * @return the lockUpdate
+     */
+    public boolean isLockUpdate() {
+        return lockUpdate;
+    }
+    /**
+     * @param lockUpdate the lockUpdate to set
+     */
+    public void setLockUpdate(boolean lockUpdate) {
+        this.lockUpdate = lockUpdate;
+    }
+    
     /**
      * @return if insert message
      */
@@ -115,13 +157,17 @@ public class EntryMessage implements Externalizable{
         return this.type != null && this.type.equals(MessageType.SYNC);
     }
     
+    
     public EntryMessage copy() {
         EntryMessage result = new EntryMessage();
         result.key=this.key;
         result.value=this.value;
+        result.oldValue=this.oldValue;
         result.type=this.type;
         result.mapUpdate=this.mapUpdate;
         result.expired=this.expired;
+        result.lockExpired=this.lockExpired;
+        result.lockUpdate=this.lockUpdate;
         return result;
     }
     
@@ -131,21 +177,28 @@ public class EntryMessage implements Externalizable{
             ClassNotFoundException {
         this.key=(EntryKey) in.readObject();
         this.value=in.readObject();
+        this.oldValue=in.readObject();
         this.type=(MessageType) in.readObject();  
         this.mapUpdate=in.readBoolean();
         this.expired=in.readBoolean();
+        this.lockExpired=in.readBoolean();
+        this.lockUpdate=in.readBoolean();
     }
     
     public void writeExternal(ObjectOutput out) throws IOException {
         out.writeObject(this.key);
         out.writeObject(this.value);
+        out.writeObject(this.oldValue);
         out.writeObject(this.type);
         out.writeBoolean(this.mapUpdate);
         out.writeBoolean(this.expired);
+        out.writeBoolean(this.lockExpired);
+        out.writeBoolean(this.lockUpdate);
     }
     
     public String toString() {
         return "EntryMessage: "+this.type + "[" + this.key + "," + this.value +
             "]{update=" + this.mapUpdate + "}";
     }
+    
 }

activemq-core/src/main/java/org/apache/activemq/group/EntryValue.java

@@ -22,20 +22,20 @@ package org.apache.activemq.group;
  *
  */
 class EntryValue<V> {
-    private Member owner;
+    private EntryKey key;
     private V value;
     
     
-    EntryValue(Member owner, V value){
+    EntryValue(EntryKey key, V value){
-        this.owner=owner;
+        this.key=key;
         this.value=value;
     }
     
     /**
      * @return the owner
      */
-    public Member getOwner() {
+    public EntryKey getKey() {
-        return this.owner;
+        return this.key;
     }
 
     /**

activemq-core/src/main/java/org/apache/activemq/group/GroupMessageListener.java

@@ -0,0 +1,34 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.activemq.group;
+
+
+/**
+ * A listener for message communication
+ *
+ */
+public interface GroupMessageListener {
+    
+    /**
+     * Called when a message is delivered to the <CODE>Group</CODE> from another member
+     * @param sender the member who sent the message
+     * @param replyId the id to use to respond to a message
+     * @param message the message object
+     */
+    void messageDelivered(Member sender, String replyId, Object message);
+}

activemq-core/src/main/java/org/apache/activemq/group/Member.java

@@ -22,6 +22,7 @@ import java.io.ObjectInput;
 import java.io.ObjectOutput;
 import javax.jms.Destination;
 import org.apache.activemq.util.IdGenerator;
+import com.sun.jndi.url.corbaname.corbanameURLContextFactory;
 
 /**
  *<P>
@@ -32,9 +33,10 @@ public class Member implements Externalizable {
     private String name;
     private String id;
     private String hostname;
-    private long timeStamp;
     private long startTime;
+    private int coordinatorWeight;
     private Destination inBoxDestination;
+    private transient long timeStamp;
     
 
     /**
@@ -94,12 +96,41 @@ public class Member implements Externalizable {
         this.inBoxDestination=dest;
     }
     
+    /**
+     * @return the timeStamp
+     */
+    long getTimeStamp() {
+        return this.timeStamp;
+    }
+
+    /**
+     * @param timeStamp the timeStamp to set
+     */
+    void setTimeStamp(long timeStamp) {
+        this.timeStamp = timeStamp;
+    }
+    /**
+     * @return the coordinatorWeight
+     */
+    public int getCoordinatorWeight() {
+        return this.coordinatorWeight;
+    }
+    /**
+     * @param coordinatorWeight the coordinatorWeight to set
+     */
+    public void setCoordinatorWeight(int coordinatorWeight) {
+        this.coordinatorWeight = coordinatorWeight;
+    }
+    
+    
+    
     public String toString() {
         return this.name+"["+this.id+"]@"+this.hostname;
     }
     
      
     public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+        this.coordinatorWeight=in.readInt();;
         this.name = in.readUTF();
         this.id = in.readUTF();
         this.hostname = in.readUTF();
@@ -108,6 +139,7 @@ public class Member implements Externalizable {
     }
 
     public void writeExternal(ObjectOutput out) throws IOException {
+        out.writeInt(this.coordinatorWeight);
         out.writeUTF(this.name != null ? this.name : "");
         out.writeUTF(this.id != null ? this.id : "");
         out.writeUTF(this.hostname != null ? this.hostname : "");
@@ -127,20 +159,4 @@ public class Member implements Externalizable {
         }
         return result;
     }
-
-    /**
-     * @return the timeStamp
-     */
-    long getTimeStamp() {
-        return this.timeStamp;
-    }
-
-    /**
-     * @param timeStamp the timeStamp to set
-     */
-    void setTimeStamp(long timeStamp) {
-        this.timeStamp = timeStamp;
-    }
-    
-    
 }

activemq-core/src/main/java/org/apache/activemq/group/MemberChangedListener.java

@@ -18,7 +18,7 @@
 package org.apache.activemq.group;
 
 /**
- * @author rajdavies
+ * A listener for membership changes to a group
  *
  */
 public interface MemberChangedListener {

activemq-core/src/main/java/org/apache/activemq/group/package.html

@@ -19,7 +19,7 @@
 </head>
 <body>
 
-Shared state and membership information between members of a remote group
+Shared state, messaging and membership information between members of a distributed  group
 
 </body>
 </html>

activemq-core/src/test/java/org/apache/activemq/group/GroupMapTest.java

@@ -34,12 +34,12 @@ public class GroupMapTest extends TestCase {
 
     /**
      * Test method for
-     * {@link org.apache.activemq.group.GroupMap#addMemberChangedListener(org.apache.activemq.group.MemberChangedListener)}.
+     * {@link org.apache.activemq.group.Group#addMemberChangedListener(org.apache.activemq.group.MemberChangedListener)}.
      * @throws Exception 
      */
     public void testAddMemberChangedListener() throws Exception {
         final AtomicInteger counter = new AtomicInteger();
-        GroupMap map1 = new GroupMap(connection1,"map1");
+        Group map1 = new Group(connection1,"map1");
         map1.addMemberChangedListener(new MemberChangedListener(){
 
             public void memberStarted(Member member) {
@@ -65,7 +65,7 @@ public class GroupMapTest extends TestCase {
             }
         }
         assertEquals(1, counter.get());
-        GroupMap map2 = new GroupMap(connection2,"map2");
+        Group map2 = new Group(connection2,"map2");
         map2.start();
         synchronized(counter) {
             if (counter.get()<2) {
@@ -76,7 +76,7 @@ public class GroupMapTest extends TestCase {
         map2.stop();
         synchronized(counter) {
             if (counter.get()>=2) {
-                counter.wait(GroupMap.DEFAULT_HEART_BEAT_INTERVAL*3);
+                counter.wait(Group.DEFAULT_HEART_BEAT_INTERVAL*3);
             }
         }
         assertEquals(1, counter.get());
@@ -85,14 +85,14 @@ public class GroupMapTest extends TestCase {
 
     /**
      * Test method for
-     * {@link org.apache.activemq.group.GroupMap#addMapChangedListener(org.apache.activemq.group.MapChangedListener)}.
+     * {@link org.apache.activemq.group.Group#addMapChangedListener(org.apache.activemq.group.MapChangedListener)}.
      * @throws Exception 
      */
     public void testAddMapChangedListener() throws Exception {
         final AtomicBoolean called1 = new AtomicBoolean();
         final AtomicBoolean called2 = new AtomicBoolean();
         
-        GroupMap map1 = new GroupMap(connection1,"map1");
+        Group map1 = new Group(connection1,"map1");
         
         map1.addMapChangedListener(new DefaultMapChangedListener() {
             public void mapInsert(Member owner,Object Key, Object Value) {
@@ -104,7 +104,7 @@ public class GroupMapTest extends TestCase {
         });
         map1.start();
         
-        GroupMap map2 = new GroupMap(connection2,"map2");
+        Group map2 = new Group(connection2,"map2");
         
         map2.addMapChangedListener(new DefaultMapChangedListener() {
             public void mapInsert(Member owner,Object Key, Object Value) {
@@ -133,12 +133,13 @@ public class GroupMapTest extends TestCase {
         map1.stop();
         map2.stop();
     }
-    
+   
-    public void testGetWriteLock() throws Exception {
+    public void testGetImplicitWriteLock() throws Exception {
-        GroupMap map1 = new GroupMap(connection1, "map1");
+        Group map1 = new Group(connection1, "map1");
         final AtomicBoolean called = new AtomicBoolean();
         map1.start();
-        GroupMap map2 = new GroupMap(connection2, "map2");
+        Group map2 = new Group(connection2, "map2");
+        map2.setAlwaysLock(true);
         map2.setMinimumGroupSize(2);
         map2.start();
         map2.put("test", "foo");
@@ -150,15 +151,84 @@ public class GroupMapTest extends TestCase {
         map1.stop();
         map2.stop();
     }
-
+    
+    public void testExpireImplicitWriteLock() throws Exception {
+        Group map1 = new Group(connection1, "map1");
+        final AtomicBoolean called = new AtomicBoolean();
+        map1.start();
+        Group map2 = new Group(connection2, "map2");
+        map2.setAlwaysLock(true);
+        map2.setLockTimeToLive(1000);
+        map2.setMinimumGroupSize(2);
+        map2.start();
+        map2.put("test", "foo");
+        try {
+            map1.put("test", "bah");
+            fail("Should have thrown an exception!");
+        } catch (GroupMapUpdateException e) {
+        }
+        Thread.sleep(2000);
+        map1.put("test", "bah");
+        map1.stop();
+        map2.stop();
+    }
+    
+    public void testExpireImplicitLockOnExit() throws Exception {
+        Group map1 = new Group(connection1, "map1");
+        final AtomicBoolean called = new AtomicBoolean();
+        map1.start();
+        Group map2 = new Group(connection2, "map2");
+        map2.setAlwaysLock(true);
+        map2.setMinimumGroupSize(2);
+        map2.start();
+        map2.put("test", "foo");
+        try {
+            map1.put("test", "bah");
+            fail("Should have thrown an exception!");
+        } catch (GroupMapUpdateException e) {
+        }
+        map2.stop();
+        map1.put("test", "bah");
+        map1.stop();
+        
+    }
+    
+    public void testGetExplicitWriteLock() throws Exception {
+        Group map1 = new Group(connection1, "map1");
+        map1.setAlwaysLock(true);
+        final AtomicBoolean called = new AtomicBoolean();
+        map1.start();
+        Group map2 = new Group(connection2, "map2");
+        map2.setAlwaysLock(true);
+        map2.setMinimumGroupSize(2);
+        map2.start();
+        map2.put("test", "foo");
+        map2.lock("test");
+        try {
+            map1.put("test", "bah");
+            fail("Should have thrown an exception!");
+        } catch (GroupMapUpdateException e) {
+        }
+        map2.unlock("test");
+        map1.lock("test");
+        try {
+            map2.lock("test");
+            fail("Should have thrown an exception!");
+        } catch (GroupMapUpdateException e) {
+        }
+        map1.stop();
+        map2.stop();
+    }
+    
+    
 
     /**
-     * Test method for {@link org.apache.activemq.group.GroupMap#clear()}.
+     * Test method for {@link org.apache.activemq.group.Group#clear()}.
      * 
      * @throws Exception
      */
     public void testClear() throws Exception {
-        GroupMap map1 = new GroupMap(connection1,"map1");
+        Group map1 = new Group(connection1,"map1");
         final AtomicBoolean called = new AtomicBoolean();
         map1.addMapChangedListener(new DefaultMapChangedListener() {
             public void mapInsert(Member owner,Object Key, Object Value) {
@@ -176,7 +246,7 @@ public class GroupMapTest extends TestCase {
             }
         });
         map1.start();
-        GroupMap map2 = new GroupMap(connection2,"map2");
+        Group map2 = new Group(connection2,"map2");
         map2.start();
         map2.put("test","foo");
         synchronized(called) {
@@ -202,12 +272,12 @@ public class GroupMapTest extends TestCase {
      * Test a new map is populated for existing values
      */
     public void testMapUpdatedOnStart() throws Exception {
-        GroupMap map1 = new GroupMap(connection1,"map1");
+        Group map1 = new Group(connection1,"map1");
         final AtomicBoolean called = new AtomicBoolean();
         
         map1.start();
         map1.put("test", "foo");
-        GroupMap map2 = new GroupMap(connection2,"map2");
+        Group map2 = new Group(connection2,"map2");
         map2.addMapChangedListener(new DefaultMapChangedListener() {
             public void mapInsert(Member owner,Object Key, Object Value) {
                 synchronized(called) {
@@ -230,9 +300,9 @@ public class GroupMapTest extends TestCase {
         map1.stop();
         map2.stop();
     }
-    
+   
     public void testContainsKey() throws Exception {
-        GroupMap map1 = new GroupMap(connection1,"map1");
+        Group map1 = new Group(connection1,"map1");
         final AtomicBoolean called = new AtomicBoolean();
         map1.addMapChangedListener(new DefaultMapChangedListener() {
             public void mapInsert(Member owner,Object Key, Object Value) {
@@ -243,7 +313,7 @@ public class GroupMapTest extends TestCase {
             }
         });
         map1.start();
-        GroupMap map2 = new GroupMap(connection2,"map2");
+        Group map2 = new Group(connection2,"map2");
         map2.start();
         map2.put("test","foo");
         synchronized(called) {
@@ -261,11 +331,11 @@ public class GroupMapTest extends TestCase {
 
     /**
      * Test method for
-     * {@link org.apache.activemq.group.GroupMap#containsValue(java.lang.Object)}.
+     * {@link org.apache.activemq.group.Group#containsValue(java.lang.Object)}.
      * @throws Exception 
      */
     public void testContainsValue() throws Exception {
-        GroupMap map1 = new GroupMap(connection1,"map1");
+        Group map1 = new Group(connection1,"map1");
         final AtomicBoolean called = new AtomicBoolean();
         map1.addMapChangedListener(new DefaultMapChangedListener() {
             public void mapInsert(Member owner,Object Key, Object Value) {
@@ -276,7 +346,7 @@ public class GroupMapTest extends TestCase {
             }
         });
         map1.start();
-        GroupMap map2 = new GroupMap(connection2,"map2");
+        Group map2 = new Group(connection2,"map2");
         map2.start();
         map2.put("test","foo");
         synchronized(called) {
@@ -299,11 +369,11 @@ public class GroupMapTest extends TestCase {
 
     /**
      * Test method for
-     * {@link org.apache.activemq.group.GroupMap#get(java.lang.Object)}.
+     * {@link org.apache.activemq.group.Group#get(java.lang.Object)}.
      * @throws Exception 
      */
     public void testGet() throws Exception {
-        GroupMap map1 = new GroupMap(connection1,"map1");
+        Group map1 = new Group(connection1,"map1");
         final AtomicBoolean called = new AtomicBoolean();
         map1.addMapChangedListener(new DefaultMapChangedListener() {
             public void mapInsert(Member owner,Object Key, Object Value) {
@@ -314,7 +384,7 @@ public class GroupMapTest extends TestCase {
             }
         });
         map1.start();
-        GroupMap map2 = new GroupMap(connection2,"map2");
+        Group map2 = new Group(connection2,"map2");
         map2.start();
         map2.put("test","foo");
         synchronized(called) {
@@ -327,15 +397,29 @@ public class GroupMapTest extends TestCase {
         map1.stop();
         map2.stop();
     }
+    
+    public void testPut() throws Exception {
+        Group map1 = new Group(connection1,"map1");
+        map1.start();
+        Group map2 = new Group(connection2,"map2");
+        map2.setMinimumGroupSize(2);
+        map2.start();
+        Object value = map1.put("foo", "blob");
+        assertNull(value);
+        value = map1.put("foo", "blah");
+        assertEquals(value, "blob");
+        map1.stop();
+        map2.stop();
+    }
 
     
     
     /**
      * Test method for
-     * {@link org.apache.activemq.group.GroupMap#remove(java.lang.Object)}.
+     * {@link org.apache.activemq.group.Group#remove(java.lang.Object)}.
      */
     public void testRemove() throws Exception{
-        GroupMap map1 = new GroupMap(connection1,"map1");
+        Group map1 = new Group(connection1,"map1");
         final AtomicBoolean called = new AtomicBoolean();
         map1.addMapChangedListener(new DefaultMapChangedListener() {
             public void mapInsert(Member owner,Object Key, Object Value) {
@@ -353,7 +437,7 @@ public class GroupMapTest extends TestCase {
             }
         });
         map1.start();
-        GroupMap map2 = new GroupMap(connection2,"map2");
+        Group map2 = new Group(connection2,"map2");
         map2.start();
         map2.put("test","foo");
         synchronized(called) {
@@ -380,7 +464,7 @@ public class GroupMapTest extends TestCase {
         final AtomicBoolean called1 = new AtomicBoolean();
         final AtomicBoolean called2 = new AtomicBoolean();
         
-        GroupMap map1 = new GroupMap(connection1,"map1");
+        Group map1 = new Group(connection1,"map1");
         map1.setTimeToLive(1000);
         map1.addMapChangedListener(new DefaultMapChangedListener() {
             public void mapRemove(Member owner, Object key, Object value,boolean expired) {        
@@ -392,7 +476,7 @@ public class GroupMapTest extends TestCase {
         });
         map1.start();
         
-        GroupMap map2 = new GroupMap(connection2,"map2");
+        Group map2 = new Group(connection2,"map2");
         
         map2.addMapChangedListener(new DefaultMapChangedListener() {
             public void mapRemove(Member owner, Object key, Object value,boolean expired) {        

activemq-core/src/test/java/org/apache/activemq/group/GroupMemberTest.java

@@ -26,27 +26,45 @@ import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.activemq.broker.BrokerService;
 
 
-public class GroupMapMemberTest extends TestCase {
+public class GroupMemberTest extends TestCase {
     protected BrokerService broker;
     protected String bindAddress = ActiveMQConnectionFactory.DEFAULT_BROKER_BIND_URL;
 
+    
+    public void testCoordinatorSelection() throws Exception{
+        Group group = new Group(null,"");
+        List<Member>list = new ArrayList<Member>();
+        final int number =10;
+        Member choosen = null;
+        for (int i =0;i< number;i++) {
+            Member m = new Member("group"+i);
+            m.setId(""+i);
+            if (number/2==i) {
+                m.setCoordinatorWeight(10);
+                choosen=m;
+            }
+            list.add(m);
+        }
+        Member c = group.selectCordinator(list);
+        assertEquals(c,choosen);
+    }
     /**
      * Test method for
-     * {@link org.apache.activemq.group.GroupMap#addMemberChangedListener(org.apache.activemq.group.MemberChangedListener)}.
+     * {@link org.apache.activemq.group.Group#addMemberChangedListener(org.apache.activemq.group.MemberChangedListener)}.
      * @throws Exception 
      */
     public void testGroup() throws Exception {
         
-        int number = 20;
+        final int number = 10;
         List<Connection>connections = new ArrayList<Connection>();
-        List<GroupMap>groupMaps = new ArrayList<GroupMap>();
+        List<Group>groupMaps = new ArrayList<Group>();
         ConnectionFactory factory = createConnectionFactory();
         for (int i =0; i < number; i++) {
             Connection connection = factory.createConnection();
             connection.start();
             connections.add(connection);
-            GroupMap map = new GroupMap(connection,"map"+i);
+            Group map = new Group(connection,"map"+i);
-            map.setHeartBeatInterval(20000);
+            map.setHeartBeatInterval(200);
             if(i ==number-1) {
                 map.setMinimumGroupSize(number);
             }
@@ -54,20 +72,56 @@ public class GroupMapMemberTest extends TestCase {
             groupMaps.add(map);
         }
         
+        int coordinatorNumber = 0;
+        for (Group map:groupMaps) {
+            if (map.isCoordinator()) {
+                coordinatorNumber++;
+            }
+        }
+        for(Group map:groupMaps) {
+            map.stop();
+        }
+        for (Connection connection:connections) {
+            connection.stop();
+        }
+        
+    }
+    
+public void XtestWeightedGroup() throws Exception {
+        
+        final int number = 10;
+        List<Connection>connections = new ArrayList<Connection>();
+        List<Group>groupMaps = new ArrayList<Group>();
+        Group last = null;
+        ConnectionFactory factory = createConnectionFactory();
+        for (int i =0; i < number; i++) {
+            Connection connection = factory.createConnection();
+            connection.start();
+            connections.add(connection);
+            Group map = new Group(connection,"map"+i);
+            map.setHeartBeatInterval(200);
+            if(i ==number-1) {
+                map.setMinimumGroupSize(number);
+                map.setCoordinatorWeight(10);
+                last=map;
+            }
+            map.start();
+            groupMaps.add(map);
+        }
+        
         int coordinator = 0;
-        for (GroupMap map:groupMaps) {
+        Group groupCoordinator = null;
+        for (Group map:groupMaps) {
             if (map.isCoordinator()) {
                 coordinator++;
+                groupCoordinator=map;
             }
         }
                
+        assertNotNull(groupCoordinator);
+        assertEquals(groupCoordinator, last);
         assertEquals(1,coordinator);
-        groupMaps.get(0).put("key", "value");
+        for(Group map:groupMaps) {
-        Thread.sleep(2000);
-        for (GroupMap map:groupMaps) {
-            assertTrue(map.get("key").equals("value"));
-        }
-        for(GroupMap map:groupMaps) {
             map.stop();
         }
         for (Connection connection:connections) {

activemq-core/src/test/java/org/apache/activemq/group/GroupMessageTest.java

@@ -0,0 +1,256 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.group;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import junit.framework.TestCase;
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+
+public class GroupMessageTest extends TestCase {
+    protected BrokerService broker;
+    protected String bindAddress = ActiveMQConnectionFactory.DEFAULT_BROKER_BIND_URL;
+
+    public void testGroupBroadcast() throws Exception {
+        final int number = 10;
+        final AtomicInteger count = new AtomicInteger();
+        List<Connection> connections = new ArrayList<Connection>();
+        List<Group> groups = new ArrayList<Group>();
+        ConnectionFactory factory = createConnectionFactory();
+        for (int i = 0; i < number; i++) {
+            Connection connection = factory.createConnection();
+            connection.start();
+            connections.add(connection);
+            Group group = new Group(connection, "group" + i);
+            group.setHeartBeatInterval(20000);
+            if (i == number - 1) {
+                group.setMinimumGroupSize(number);
+            }
+            group.start();
+            groups.add(group);
+            group.addGroupMessageListener(new GroupMessageListener() {
+                public void messageDelivered(Member sender, String replyId,
+                        Object message) {
+                    synchronized (count) {
+                        if (count.incrementAndGet() == number) {
+                            count.notifyAll();
+                        }
+                    }
+                }
+            });
+        }
+        groups.get(0).broadcastMessage("hello");
+        synchronized (count) {
+            if (count.get() < number) {
+                count.wait(5000);
+            }
+        }
+        assertEquals(number, count.get());
+        for (Group map : groups) {
+            map.stop();
+        }
+        for (Connection connection : connections) {
+            connection.stop();
+        }
+    }
+
+    public void testsendMessage() throws Exception {
+        final int number = 10;
+        final AtomicInteger count = new AtomicInteger();
+        List<Connection> connections = new ArrayList<Connection>();
+        List<Group> groups = new ArrayList<Group>();
+        ConnectionFactory factory = createConnectionFactory();
+        for (int i = 0; i < number; i++) {
+            Connection connection = factory.createConnection();
+            connection.start();
+            connections.add(connection);
+            Group group = new Group(connection, "group" + i);
+            group.setHeartBeatInterval(20000);
+            if (i == number - 1) {
+                group.setMinimumGroupSize(number);
+            }
+            group.start();
+            groups.add(group);
+            group.addGroupMessageListener(new GroupMessageListener() {
+                public void messageDelivered(Member sender, String replyId,
+                        Object message) {
+                    synchronized (count) {
+                        count.incrementAndGet();
+                        count.notifyAll();
+                    }
+                }
+            });
+        }
+        groups.get(0).sendMessage("hello");
+        synchronized (count) {
+            if (count.get() == 0) {
+                count.wait(5000);
+            }
+        }
+        // wait a while to check that only one got it
+        Thread.sleep(2000);
+        assertEquals(1, count.get());
+        for (Group map : groups) {
+            map.stop();
+        }
+        for (Connection connection : connections) {
+            connection.stop();
+        }
+    }
+
+    public void testSendToSingleMember() throws Exception {
+        ConnectionFactory factory = createConnectionFactory();
+        Connection connection1 = factory.createConnection();
+        Connection connection2 = factory.createConnection();
+        connection1.start();
+        connection2.start();
+        Group group1 = new Group(connection1, "group1");
+        final AtomicBoolean called = new AtomicBoolean();
+        group1.addGroupMessageListener(new GroupMessageListener() {
+            public void messageDelivered(Member sender, String replyId,
+                    Object message) {
+                synchronized (called) {
+                    called.set(true);
+                    called.notifyAll();
+                }
+            }
+        });
+        group1.start();
+        Group group2 = new Group(connection2, "group2");
+        group2.setMinimumGroupSize(2);
+        group2.start();
+        Member member1 = group2.getMemberByName("group1");
+        group2.sendMessage(member1, "hello");
+        synchronized (called) {
+            if (!called.get()) {
+                called.wait(5000);
+            }
+        }
+        assertTrue(called.get());
+        group1.stop();
+        group2.stop();
+        connection1.close();
+        connection2.close();
+    }
+
+    public void testSendRequestReply() throws Exception {
+        ConnectionFactory factory = createConnectionFactory();
+        Connection connection1 = factory.createConnection();
+        Connection connection2 = factory.createConnection();
+        connection1.start();
+        connection2.start();
+        final int number = 1000;
+        final AtomicInteger requestCount = new AtomicInteger();
+        final AtomicInteger replyCount = new AtomicInteger();
+        final List<String> requests = new ArrayList<String>();
+        final List<String> replies = new ArrayList<String>();
+        for (int i = 0; i < number; i++) {
+            requests.add("request" + i);
+            replies.add("reply" + i);
+        }
+        final Group group1 = new Group(connection1, "group1");
+        final AtomicBoolean finished = new AtomicBoolean();
+        group1.addGroupMessageListener(new GroupMessageListener() {
+            public void messageDelivered(Member sender, String replyId,
+                    Object message) {
+                if (!replies.isEmpty()) {
+                    String reply = replies.remove(0);
+                    try {
+                        group1.sendMessageResponse(sender, replyId, reply);
+                    } catch (JMSException e) {
+                        e.printStackTrace();
+                    }
+                } 
+            }
+        });
+        group1.start();
+        final Group group2 = new Group(connection2, "group2");
+        group2.setMinimumGroupSize(2);
+        group2.addGroupMessageListener(new GroupMessageListener() {
+            public void messageDelivered(Member sender, String replyId,
+                    Object message) {
+                if (!requests.isEmpty()) {
+                    String request = requests.remove(0);
+                    try {
+                        group2.sendMessage(sender, request);
+                    } catch (JMSException e) {
+                        e.printStackTrace();
+                    }
+                }else {
+                    synchronized (finished) {
+                        finished.set(true);
+                        finished.notifyAll();
+                    }
+                }
+            }
+        });
+        group2.start();
+        Member member1 = group2.getMemberByName("group1");
+        group2.sendMessage(member1, requests.remove(0));
+        synchronized (finished) {
+            if (!finished.get()) {
+                finished.wait(10000);
+            }
+        }
+        assertTrue(finished.get());
+        group1.stop();
+        group2.stop();
+        connection1.close();
+        connection2.close();
+    }
+
+    protected void setUp() throws Exception {
+        if (broker == null) {
+            broker = createBroker();
+        }
+        super.setUp();
+    }
+
+    protected void tearDown() throws Exception {
+        super.tearDown();
+        if (broker != null) {
+            broker.stop();
+        }
+    }
+
+    protected ActiveMQConnectionFactory createConnectionFactory()
+            throws Exception {
+        ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(
+                ActiveMQConnection.DEFAULT_BROKER_URL);
+        return cf;
+    }
+
+    protected BrokerService createBroker() throws Exception {
+        BrokerService answer = new BrokerService();
+        configureBroker(answer);
+        answer.start();
+        return answer;
+    }
+
+    protected void configureBroker(BrokerService answer) throws Exception {
+        answer.setPersistent(false);
+        answer.addConnector(bindAddress);
+        answer.setDeleteAllMessagesOnStartup(true);
+    }
+}