Reconnect with channelName instead channelId (#85)

#82 Bitfinex reconnect loop

service-netty/src/main/java/info/bitrich/xchangestream/service/netty/NettyStreamingService.java

@@ -32,8 +32,15 @@ public abstract class NettyStreamingService<T> {
     private static final Logger LOG = LoggerFactory.getLogger(NettyStreamingService.class);
 
     private class Subscription {
-        ObservableEmitter<T> emitter;
-        Object[] args;
+        final ObservableEmitter<T> emitter;
+        final String channelName;
+        final Object[] args;
+
+        public Subscription(ObservableEmitter<T> emitter, String channelName, Object[] args) {
+            this.emitter = emitter;
+            this.channelName = channelName;
+            this.args = args;
+        }
     }
 
     private final int maxFramePayloadLength;
@@ -188,9 +195,7 @@ public Observable<T> subscribeChannel(String channelName, Object... args) {
             }
 
             if (!channels.containsKey(channelId)) {
-                Subscription newSubscription = new Subscription();
-                newSubscription.args = args;
-                newSubscription.emitter = e;
+                Subscription newSubscription = new Subscription(e, channelName, args);
                 channels.put(channelId, newSubscription);
                 try {
                     sendMessage(getSubscribeMessage(channelName, args));
@@ -207,11 +212,12 @@ public Observable<T> subscribeChannel(String channelName, Object... args) {
     }
 
     public void resubscribeChannels() {
-        for (String channelName : channels.keySet()) {
+        for (String channelId : channels.keySet()) {
             try {
-                sendMessage(getSubscribeMessage(channelName, channels.get(channelName).args));
+                Subscription subscription = channels.get(channelId);
+                sendMessage(getSubscribeMessage(subscription.channelName, subscription.args));
             } catch (IOException e) {
-                LOG.error("Failed to reconnect channel: {}", channelName);
+                LOG.error("Failed to reconnect channel: {}", channelId);
             }
         }
     }