Initial commit for user_id feature

Author: Anya Zenkina

src/main/java/com/pusher/client/channel/EventMetadata.java

@@ -0,0 +1,15 @@
+package com.pusher.client.channel;
+
+public class EventMetadata {
+    String userId;
+
+    public String getUserId() {
+        return userId;
+    }
+
+
+    public void setUserId(String value) {
+        this.userId = value;
+    }
+
+}

src/main/java/com/pusher/client/channel/PresenceChannelEventListener.java

@@ -39,4 +39,6 @@ public interface PresenceChannelEventListener extends PrivateChannelEventListene
      *            The user who unsubscribed.
      */
     void userUnsubscribed(String channelName, User user);
+
+    void onEventWithMetadata(String channelName, String eventName, String data, EventMetadata metadata);
 }

src/main/java/com/pusher/client/channel/impl/ChannelImpl.java

@@ -23,6 +23,11 @@ public class ChannelImpl implements InternalChannel {
     private final Map<String, Set<SubscriptionEventListener>> eventNameToListenerMap = new HashMap<String, Set<SubscriptionEventListener>>();
     protected volatile ChannelState state = ChannelState.INITIAL;
     private ChannelEventListener eventListener;
+
+    protected Factory getFactory() {
+        return factory;
+    }
+
     private final Factory factory;
     private final Object lock = new Object();
 
@@ -190,7 +195,7 @@ public String toString() {
     }
 
     @SuppressWarnings("unchecked")
-    private String extractDataFrom(final String message) {
+    protected String extractDataFrom(final String message) {
         final Map<Object, Object> jsonObject = GSON.fromJson(message, Map.class);
         return (String)jsonObject.get("data");
     }
@@ -199,7 +204,7 @@ protected String[] getDisallowedNameExpressions() {
         return new String[] { "^private-.*", "^presence-.*" };
     }
 
-    private void validateArguments(final String eventName, final SubscriptionEventListener listener) {
+    protected void validateArguments(final String eventName, final SubscriptionEventListener listener) {
 
         if (eventName == null) {
             throw new IllegalArgumentException("Cannot bind or unbind to channel " + name + " with a null event name");

src/main/java/com/pusher/client/channel/impl/PresenceChannelImpl.java

@@ -5,6 +5,8 @@
 import com.pusher.client.AuthorizationFailureException;
 import com.pusher.client.Authorizer;
 import com.pusher.client.channel.ChannelEventListener;
+import com.pusher.client.channel.ChannelState;
+import com.pusher.client.channel.EventMetadata;
 import com.pusher.client.channel.PresenceChannel;
 import com.pusher.client.channel.PresenceChannelEventListener;
 import com.pusher.client.channel.SubscriptionEventListener;
@@ -13,6 +15,8 @@
 import com.pusher.client.util.Factory;
 
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -51,7 +55,47 @@ public User getMe() {
     @Override
     public void onMessage(final String event, final String message) {
 
-        super.onMessage(event, message);
+        if (event.equals(SUBSCRIPTION_SUCCESS_EVENT)) {
+            updateState(ChannelState.SUBSCRIBED);
+        }
+        else {
+            final Set<PresenceChannelEventListener> listeners;
+            synchronized (lock) {
+                final Set<PresenceChannelEventListener> sharedListeners = eventNameToListenerMap.get(event);
+                if (sharedListeners != null) {
+                    listeners = new HashSet<PresenceChannelEventListener>(sharedListeners);
+                }
+                else {
+                    listeners = null;
+                }
+            }
+
+            if (listeners != null) {
+                for (final PresenceChannelEventListener listener : listeners) {
+                    final String data = extractDataFrom(message);
+                    final EventMetadata metadata = new EventMetadata();
+
+                    // extracting user_id if client-event
+                    if (event.startsWith(CLIENT_EVENT_PREFIX)) {
+
+                        // extracting user_id from message
+                        String userId = extractUserIdStringFrom(message);
+
+                        metadata.setUserId(userId);
+
+                    }
+                        getFactory().queueOnEventThread(new Runnable() {
+                        @Override
+                        public void run() {
+                            listener.onEvent(name, event, data);
+
+                            listener.onEventWithMetadata(name, event, data, metadata);
+                        }
+                    });
+                }
+            }
+        }
+
 
         if (event.equals(SUBSCRIPTION_SUCCESS_EVENT)) {
             handleSubscriptionSuccessfulMessage(message);
@@ -96,6 +140,9 @@ public String toSubscribeMessage() {
         }
     }
 
+    private final Object lock = new Object();
+    private final Map<String, Set<PresenceChannelEventListener>> eventNameToListenerMap = new HashMap<String, Set<PresenceChannelEventListener>>();
+
     @Override
     public void bind(final String eventName, final SubscriptionEventListener listener) {
 
@@ -104,7 +151,31 @@ public void bind(final String eventName, final SubscriptionEventListener listene
                     "Only instances of PresenceChannelEventListener can be bound to a presence channel");
         }
 
-        super.bind(eventName, listener);
+                validateArguments(eventName, listener);
+
+        synchronized (lock) {
+            Set<PresenceChannelEventListener> listeners = eventNameToListenerMap.get(eventName);
+            if (listeners == null) {
+                listeners = new HashSet<PresenceChannelEventListener>();
+                eventNameToListenerMap.put(eventName, listeners);
+            }
+            listeners.add((PresenceChannelEventListener)listener);
+        }
+    }
+
+    @Override
+    public void unbind(final String eventName, final SubscriptionEventListener listener) {
+        validateArguments(eventName, listener);
+
+        synchronized (lock) {
+            final Set<PresenceChannelEventListener> listeners = eventNameToListenerMap.get(eventName);
+            if (listeners != null) {
+                listeners.remove(listener);
+                if (listeners.isEmpty()) {
+                    eventNameToListenerMap.remove(eventName);
+                }
+            }
+        }
     }
 
     @Override
@@ -177,7 +248,12 @@ private void handleMemberRemovedEvent(final String message) {
     @SuppressWarnings("rawtypes")
     private static String extractDataStringFrom(final String message) {
         final Map jsonObject = GSON.fromJson(message, Map.class);
-        return  (String) jsonObject.get("data");
+        return (String) jsonObject.get("data");
+    }
+
+    private static String extractUserIdStringFrom(final String message) {
+        final Map jsonObject = GSON.fromJson(message, Map.class);
+        return (String) jsonObject.get("user_id");
     }
 
     @SuppressWarnings("rawtypes")

src/main/java/com/pusher/client/channel/impl/PrivateChannelImpl.java

@@ -19,7 +19,7 @@
 public class PrivateChannelImpl extends ChannelImpl implements PrivateChannel {
 
     private static final Gson GSON = new Gson();
-    private static final String CLIENT_EVENT_PREFIX = "client-";
+    protected static final String CLIENT_EVENT_PREFIX = "client-";
     private final InternalConnection connection;
     private final Authorizer authorizer;
 

src/main/java/com/pusher/client/example/PresenceChannelExampleApp.java

@@ -4,6 +4,7 @@
 
 import com.pusher.client.Pusher;
 import com.pusher.client.PusherOptions;
+import com.pusher.client.channel.EventMetadata;
 import com.pusher.client.channel.PresenceChannel;
 import com.pusher.client.channel.PresenceChannelEventListener;
 import com.pusher.client.channel.User;
@@ -99,6 +100,13 @@ public void onEvent(final String channelName, final String eventName, final Stri
                 data));
     }
 
+    @Override
+    public void onEventWithMetadata(final String channelName, final String eventName, final String data, final EventMetadata metadata) {
+
+        System.out.println(String.format("Received event [%s] on channel [%s] with data [%s] with metadata user_id: [%s]", eventName, channelName,
+                data, metadata.getUserId()));
+    }
+
     @Override
     public void onSubscriptionSucceeded(final String channelName) {
 

src/test/java/com/pusher/client/channel/impl/PresenceChannelImplTest.java

@@ -20,6 +20,7 @@
 
 import com.pusher.client.channel.ChannelEventListener;
 import com.pusher.client.channel.ChannelState;
+import com.pusher.client.channel.EventMetadata;
 import com.pusher.client.channel.PresenceChannelEventListener;
 import com.pusher.client.channel.PrivateChannelEventListener;
 import com.pusher.client.channel.User;
@@ -69,6 +70,23 @@ public void testIsSubscribedMethod(){
         assertTrue(channel.isSubscribed());
     }
 
+    @Override
+    @Test
+    public void testDataIsExtractedFromMessageAndPassedToSingleListener() {
+        final ArgumentCaptor<EventMetadata> argument = ArgumentCaptor.forClass(EventMetadata.class);
+
+        final String eventName = "client-my-event";
+        PresenceChannelEventListener mockListener = getEventListener();
+
+        channel = newInstance(getChannelName());
+        channel.bind(eventName, mockListener);
+        channel.onMessage(eventName, "{\"event\":\"client-my-event\",\"data\":\"{\\\"fish\\\":\\\"chips\\\"}\",\"user_id\":\"Kuzya\"}");
+
+        verify(mockListener).onEventWithMetadata(eq(getChannelName()), eq(eventName), eq("{\"fish\":\"chips\"}"), argument.capture());
+
+        assertEquals("Kuzya", argument.getValue().getUserId());
+    }
+
     @Test
     @Override
     @SuppressWarnings({ "rawtypes", "unchecked" })
@@ -225,7 +243,7 @@ protected String getChannelName() {
     }
 
     @Override
-    protected ChannelEventListener getEventListener() {
+    protected PresenceChannelEventListener getEventListener() {
         return mock(PresenceChannelEventListener.class);
     }