Pilot implementation for Realtime event based API's - ConnectionStateChange

README.md

@@ -25,6 +25,92 @@ or visit our
 [Support and Help](https://www.ably.io/support)
 site to discuss privately.
 
+## Running the example
+
+* Clone the repo
+* cd to `example` folder
+* run `flutter packages get` to install dependencies
+* `flutter run` will start the application on connected android / iOS device
+
+
+## Using the Realtime API
+
+##### Import the package
+
+```dart
+import 'package:ably_flutter_plugin/ably.dart' as ably;
+```
+
+##### Create an Ably instance
+
+```dart
+final ably.Ably ablyPlugin = ably.Ably();
+```
+
+##### create a ClientOptions
+
+```dart
+final clientOptions = ably.ClientOptions.fromKey("<YOUR APP KEY>");
+clientOptions.environment = '<your env>';       //optional
+clientOptions.logLevel = ably.LogLevel.verbose; //optional
+```
+
+##### Rest API
+
+```dart
+ably.Rest rest = await ablyPlugin.createRest(options: clientOptions);
+```
+
+Getting a channel instance
+
+```dart
+ably.RestChannel channel = rest.channels.get('test');
+```
+
+Publish rest messages
+
+```dart
+//passing both name and data
+await channel.publish(name: "Hello", data: "Ably");
+//passing just name
+await channel.publish(name: "Hello");
+//passing just data
+await channel.publish(data: "Ably");
+//publishing an empty message
+await channel.publish();
+```
+
+##### Realtime API
+
+```dart
+ably.Realtime realtime = await ablyPlugin.createRealtime(options: clientOptions);
+```
+
+Listen to connection state change event
+
+```dart
+realtime.connection.on().listen((ably.ConnectionStateChange stateChange) async {
+  print('Realtime connection state changed: ${stateChange.event}');
+  setState(() { _realtimeConnectionState = stateChange.current; });
+});
+```
+
+Listening to a particular event: `connected`
+
+```dart
+realtime.connection.on(ably.ConnectionEvent.connected).listen((ably.ConnectionStateChange stateChange) async {
+  print('Realtime connection state changed: ${stateChange.event}');
+  setState(() { _realtimeConnectionState = stateChange.current; });
+});
+```
+
+Connect and disconnect to a realtime instance
+
+```dart
+realtime.connect();     //connect to realtime
+realtime.disconnect();  //disconnect from realtime
+```
+
 ## Contributing
 
 We have some [Developer Notes](DeveloperNotes.md), but we're still learning too so they'll only help you so far, in fact there's probably a lot you can teach us!

android/src/main/java/io/ably/flutter/plugin/AblyEventStreamHandler.java

@@ -0,0 +1,117 @@
+package io.ably.flutter.plugin;
+
+import android.os.Handler;
+import android.os.Looper;
+
+import io.ably.lib.realtime.ChannelStateListener;
+import io.ably.lib.realtime.ConnectionStateListener;
+import io.flutter.plugin.common.EventChannel;
+
+
+/**
+ * Dart side can listen to Event Streams by pushing data to eventSink available in onListen method.
+ * Event listening can be cancelled when stream subscription is cancelled on dart side
+ *
+ * ref: https://api.flutter.dev/javadoc/io/flutter/plugin/common/EventChannel.StreamHandler.html
+ * */
+public class AblyEventStreamHandler implements EventChannel.StreamHandler {
+    private final AblyMethodCallHandler methodCallHandler;
+
+    /**
+     * Constructor requiring methodCallHandler, as it is a singleton and has all instances stored
+     * Event listening can be started on an instance picked from the stored instances
+     * */
+    AblyEventStreamHandler(AblyMethodCallHandler methodCallHandler){
+        this.methodCallHandler = methodCallHandler;
+    }
+
+    /**
+     * Refer to the comments on AblyMethodCallHandler.MethodResultWrapper
+     * on why this customized EventSink is required
+     * */
+    private static class MainThreadEventSink implements EventChannel.EventSink {
+        private EventChannel.EventSink eventSink;
+        private Handler handler;
+
+        MainThreadEventSink(EventChannel.EventSink eventSink) {
+            this.eventSink = eventSink;
+            handler = new Handler(Looper.getMainLooper());
+        }
+
+        @Override
+        public void success(final Object o) {
+            handler.post(() -> eventSink.success(o));   //lambda for new Runnable
+        }
+
+        @Override
+        public void error(final String s, final String s1, final Object o) {
+            handler.post(() -> eventSink.error(s, s1, o));
+        }
+
+        @Override
+        public void endOfStream() {
+            //TODO work on this if required, or remove this TODO once all features are covered
+        }
+    }
+
+    // Listeners
+    private PluginConnectionStateListener connectionStateListener;
+    private PluginChannelStateListener channelStateListener;
+
+    private class Listener{
+        EventChannel.EventSink eventSink;
+        Listener(EventChannel.EventSink eventSink){ this.eventSink = eventSink; }
+    }
+
+    private class PluginConnectionStateListener extends Listener implements ConnectionStateListener {
+        PluginConnectionStateListener(EventChannel.EventSink eventSink){super(eventSink);}
+        public void onConnectionStateChanged(ConnectionStateChange stateChange){
+            eventSink.success(stateChange);
+        }
+    }
+
+    private class PluginChannelStateListener extends Listener implements ChannelStateListener {
+        PluginChannelStateListener(EventChannel.EventSink eventSink){super(eventSink);}
+        public void onChannelStateChanged(io.ably.lib.realtime.ChannelStateListener.ChannelStateChange stateChange){
+            eventSink.success(stateChange);
+        }
+    }
+
+    @Override
+    public void onListen(Object object, EventChannel.EventSink uiThreadEventSink) {
+        MainThreadEventSink eventSink = new MainThreadEventSink(uiThreadEventSink);
+        methodCallHandler.<AblyFlutterMessage<String>>ablyDo((AblyFlutterMessage)object, (ablyLibrary, message) -> {
+            String eventName = message.message;
+            switch(eventName) {
+                case "realtime_onConnectionStateChanged":
+                    connectionStateListener = new PluginConnectionStateListener(eventSink);
+                    ablyLibrary.getRealtime(message.handle).connection.on(connectionStateListener);
+                    return;
+                case "realtime_onChannelStateChanged":
+                    // channelStateListener = new PluginChannelStateListener(eventSink);
+                    // ablyLibrary.getRealtime(message.handle).connection.on(channelStateListener);
+                    // return;
+                default:
+                    eventSink.error("unhandled event", null, null);
+            }
+        });
+    }
+
+    @Override
+    public void onCancel(Object object) {
+        if(object==null){
+            System.out.println("Cannot process null input on cancel");
+            return;
+        }
+        methodCallHandler.<AblyFlutterMessage<String>>ablyDo((AblyFlutterMessage)object, (ablyLibrary, message) -> {
+            String eventName = message.message;
+            switch (eventName) {
+                case "realtime_onConnectionStateChanged":
+                    ablyLibrary.getRealtime(message.handle).connection.off(connectionStateListener);
+                case "realtime_onChannelStateChanged":
+                    // ablyLibrary.getRealtime(handle).connection.off(connectionStateListener);
+            }
+        });
+    }
+
+}

android/src/main/java/io/ably/flutter/plugin/AblyFlutterPlugin.java

@@ -2,7 +2,9 @@
 
 import androidx.annotation.NonNull;
 
+import app.loup.streams_channel.StreamsChannel;
 import io.flutter.embedding.engine.plugins.FlutterPlugin;
+import io.flutter.plugin.common.BinaryMessenger;
 import io.flutter.plugin.common.MethodChannel;
 import io.flutter.plugin.common.MethodCodec;
 import io.flutter.plugin.common.PluginRegistry.Registrar;
@@ -15,8 +17,7 @@ private static MethodCodec createCodec() {
 
     @Override
     public void onAttachedToEngine(@NonNull FlutterPluginBinding flutterPluginBinding) {
-        final MethodChannel channel = new MethodChannel(flutterPluginBinding.getBinaryMessenger(), "ably_flutter_plugin", createCodec());
-        channel.setMethodCallHandler(AblyMethodCallHandler.getInstance());
+        setupChannels(flutterPluginBinding.getBinaryMessenger());
     }
 
     // This static function is optional and equivalent to onAttachedToEngine. It supports the old
@@ -29,8 +30,19 @@ public void onAttachedToEngine(@NonNull FlutterPluginBinding flutterPluginBindin
     // depending on the user's project. onAttachedToEngine or registerWith must both be defined
     // in the same class.
     public static void registerWith(Registrar registrar) {
-        final MethodChannel channel = new MethodChannel(registrar.messenger(), "ably_flutter_plugin", createCodec());
-        channel.setMethodCallHandler(AblyMethodCallHandler.getInstance());
+        AblyFlutterPlugin.setupChannels(registrar.messenger());
+    }
+
+    private static void setupChannels(BinaryMessenger messenger){
+        MethodCodec codec = createCodec();
+        AblyMethodCallHandler methodCallHandler = AblyMethodCallHandler.getInstance();
+
+        final MethodChannel channel = new MethodChannel(messenger, "io.ably.flutter.plugin", codec);
+        channel.setMethodCallHandler(methodCallHandler);
+
+        final StreamsChannel streamsChannel = new StreamsChannel(messenger, "io.ably.flutter.stream", codec);
+        streamsChannel.setStreamHandlerFactory(arguments -> new AblyEventStreamHandler(methodCallHandler));
+
     }
 
     @Override

android/src/main/java/io/ably/flutter/plugin/AblyLibrary.java

@@ -15,7 +15,6 @@ class AblyLibrary {
     //    and as per this answer https://stackoverflow.com/a/31413003
     private final LongSparseArray<AblyRest> _restInstances = new LongSparseArray<>();
     private final LongSparseArray<AblyRealtime> _realtimeInstances = new LongSparseArray<>();
-    private final LongSparseArray<AblyRealtimeConnectionListener> _connectionListeners = new LongSparseArray<>();
 
     private void assertNotDisposed() {
         if (_disposed) {
@@ -51,21 +50,6 @@ AblyRest getRest(final long handle){
         return _restInstances.get(handle);
     }
 
-    long createConnectionListener(final AblyRealtime realtime) {
-        assertNotDisposed();
-
-        final AblyRealtimeConnectionListener listener = new AblyRealtimeConnectionListener();
-        realtime.connection.on(listener);
-        _connectionListeners.put(_nextHandle, listener);
-        return _nextHandle++;
-    }
-
-    AblyRealtimeConnectionListener getConnectionListener(final long handle) {
-        assertNotDisposed();
-
-        return _connectionListeners.get(handle);
-    }
-
     void dispose() {
         assertNotDisposed();