Implement streaming data into Flutter

example/lib/main.dart

@@ -42,6 +42,10 @@ class MyAppState extends State<MyApp> {
     await Permission.notification.request();
 
     await MoxplatformPlugin.notifications.createNotificationChannel("Test notification channel", channelId, false);
+
+    MoxplatformPlugin.notifications.getEventStream().listen((event) {
+      print('NotificationEvent(type: ${event.type}, jid: ${event.jid}, payload: ${event.payload})');
+    });
   }
 
   @override

packages/moxplatform_android/android/src/main/java/me/polynom/moxplatform_android/Api.java

@@ -57,6 +57,18 @@ public class Api {
     return errorList;
   }
 
+  public enum NotificationEventType {
+    MARK_AS_READ(0),
+    REPLY(1),
+    OPEN(2);
+
+    final int index;
+
+    private NotificationEventType(final int index) {
+      this.index = index;
+    }
+  }
+
   /** Generated class from Pigeon that represents data sent in messages. */
   public static final class NotificationMessageContent {
     /** The textual body of the message. */
@@ -441,6 +453,107 @@ public class Api {
     }
   }
 
+  /** Generated class from Pigeon that represents data sent in messages. */
+  public static final class NotificationEvent {
+    /** The JID the notification was for. */
+    private @NonNull String jid;
+
+    public @NonNull String getJid() {
+      return jid;
+    }
+
+    public void setJid(@NonNull String setterArg) {
+      if (setterArg == null) {
+        throw new IllegalStateException("Nonnull field \"jid\" is null.");
+      }
+      this.jid = setterArg;
+    }
+
+    /** The type of event. */
+    private @NonNull NotificationEventType type;
+
+    public @NonNull NotificationEventType getType() {
+      return type;
+    }
+
+    public void setType(@NonNull NotificationEventType setterArg) {
+      if (setterArg == null) {
+        throw new IllegalStateException("Nonnull field \"type\" is null.");
+      }
+      this.type = setterArg;
+    }
+
+    /**
+     * An optional payload.
+     * - type == NotificationType.reply: The reply message text.
+     * Otherwise: undefined.
+     */
+    private @Nullable String payload;
+
+    public @Nullable String getPayload() {
+      return payload;
+    }
+
+    public void setPayload(@Nullable String setterArg) {
+      this.payload = setterArg;
+    }
+
+    /** Constructor is non-public to enforce null safety; use Builder. */
+    NotificationEvent() {}
+
+    public static final class Builder {
+
+      private @Nullable String jid;
+
+      public @NonNull Builder setJid(@NonNull String setterArg) {
+        this.jid = setterArg;
+        return this;
+      }
+
+      private @Nullable NotificationEventType type;
+
+      public @NonNull Builder setType(@NonNull NotificationEventType setterArg) {
+        this.type = setterArg;
+        return this;
+      }
+
+      private @Nullable String payload;
+
+      public @NonNull Builder setPayload(@Nullable String setterArg) {
+        this.payload = setterArg;
+        return this;
+      }
+
+      public @NonNull NotificationEvent build() {
+        NotificationEvent pigeonReturn = new NotificationEvent();
+        pigeonReturn.setJid(jid);
+        pigeonReturn.setType(type);
+        pigeonReturn.setPayload(payload);
+        return pigeonReturn;
+      }
+    }
+
+    @NonNull
+    ArrayList<Object> toList() {
+      ArrayList<Object> toListResult = new ArrayList<Object>(3);
+      toListResult.add(jid);
+      toListResult.add(type == null ? null : type.index);
+      toListResult.add(payload);
+      return toListResult;
+    }
+
+    static @NonNull NotificationEvent fromList(@NonNull ArrayList<Object> list) {
+      NotificationEvent pigeonResult = new NotificationEvent();
+      Object jid = list.get(0);
+      pigeonResult.setJid((String) jid);
+      Object type = list.get(1);
+      pigeonResult.setType(type == null ? null : NotificationEventType.values()[(int) type]);
+      Object payload = list.get(2);
+      pigeonResult.setPayload((String) payload);
+      return pigeonResult;
+    }
+  }
+
   private static class MoxplatformApiCodec extends StandardMessageCodec {
     public static final MoxplatformApiCodec INSTANCE = new MoxplatformApiCodec();
 
@@ -452,8 +565,10 @@ public class Api {
         case (byte) 128:
           return MessagingNotification.fromList((ArrayList<Object>) readValue(buffer));
         case (byte) 129:
-          return NotificationMessage.fromList((ArrayList<Object>) readValue(buffer));
+          return NotificationEvent.fromList((ArrayList<Object>) readValue(buffer));
         case (byte) 130:
+          return NotificationMessage.fromList((ArrayList<Object>) readValue(buffer));
+        case (byte) 131:
           return NotificationMessageContent.fromList((ArrayList<Object>) readValue(buffer));
         default:
           return super.readValueOfType(type, buffer);
@@ -465,11 +580,14 @@ public class Api {
       if (value instanceof MessagingNotification) {
         stream.write(128);
         writeValue(stream, ((MessagingNotification) value).toList());
-      } else if (value instanceof NotificationMessage) {
+      } else if (value instanceof NotificationEvent) {
         stream.write(129);
+        writeValue(stream, ((NotificationEvent) value).toList());
+      } else if (value instanceof NotificationMessage) {
+        stream.write(130);
         writeValue(stream, ((NotificationMessage) value).toList());
       } else if (value instanceof NotificationMessageContent) {
-        stream.write(130);
+        stream.write(131);
         writeValue(stream, ((NotificationMessageContent) value).toList());
       } else {
         super.writeValue(stream, value);
@@ -490,6 +608,8 @@ public class Api {
     @NonNull 
     String getCacheDataPath();
 
+    void eventStub(@NonNull NotificationEvent event);
+
     /** The codec used by MoxplatformApi. */
     static @NonNull MessageCodec<Object> getCodec() {
       return MoxplatformApiCodec.INSTANCE;
@@ -580,6 +700,30 @@ public class Api {
                   String output = api.getCacheDataPath();
                   wrapped.add(0, output);
                 }
+ catch (Throwable exception) {
+                  ArrayList<Object> wrappedError = wrapError(exception);
+                  wrapped = wrappedError;
+                }
+                reply.reply(wrapped);
+              });
+        } else {
+          channel.setMessageHandler(null);
+        }
+      }
+      {
+        BasicMessageChannel<Object> channel =
+            new BasicMessageChannel<>(
+                binaryMessenger, "dev.flutter.pigeon.moxplatform_platform_interface.MoxplatformApi.eventStub", getCodec());
+        if (api != null) {
+          channel.setMessageHandler(
+              (message, reply) -> {
+                ArrayList<Object> wrapped = new ArrayList<Object>();
+                ArrayList<Object> args = (ArrayList<Object>) message;
+                NotificationEvent eventArg = (NotificationEvent) args.get(0);
+                try {
+                  api.eventStub(eventArg);
+                  wrapped.add(0, null);
+                }
  catch (Throwable exception) {
                   ArrayList<Object> wrappedError = wrapError(exception);
                   wrapped = wrappedError;

packages/moxplatform_android/android/src/main/java/me/polynom/moxplatform_android/Constants.kt

@@ -6,6 +6,7 @@ const val TAG = "Moxplatform"
 // The size of the buffer to hashing, encryption, and decryption in bytes.
 const val BUFFER_SIZE = 8096
 
+const val REPLY_ACTION = "reply";
 // The data key for text entered in the notification's reply field
 const val REPLY_TEXT_KEY = "key_reply_text"
 

packages/moxplatform_android/android/src/main/java/me/polynom/moxplatform_android/MoxplatformAndroidPlugin.java

@@ -25,6 +25,9 @@ import java.util.List;
 import io.flutter.embedding.engine.plugins.FlutterPlugin;
 import io.flutter.embedding.engine.plugins.service.ServiceAware;
 import io.flutter.embedding.engine.plugins.service.ServicePluginBinding;
+import io.flutter.plugin.common.EventChannel;
+import io.flutter.plugin.common.EventChannel.EventSink;
+import io.flutter.plugin.common.EventChannel.StreamHandler;
 import io.flutter.plugin.common.MethodCall;
 import io.flutter.plugin.common.MethodChannel;
 import io.flutter.plugin.common.MethodChannel.MethodCallHandler;
@@ -34,7 +37,7 @@ import io.flutter.plugin.common.JSONMethodCodec;
 import kotlin.Unit;
 import kotlin.jvm.functions.Function1;
 
-public class MoxplatformAndroidPlugin extends BroadcastReceiver implements FlutterPlugin, MethodCallHandler, ServiceAware, MoxplatformApi {
+  public class MoxplatformAndroidPlugin extends BroadcastReceiver implements FlutterPlugin, MethodCallHandler, EventChannel.StreamHandler, ServiceAware, MoxplatformApi {
   public static final String entrypointKey = "entrypoint_handle";
   public static final String extraDataKey = "extra_data";
   private static final String autoStartAtBootKey = "auto_start_at_boot";
@@ -46,7 +49,10 @@ public class MoxplatformAndroidPlugin extends BroadcastReceiver implements Flutt
   private static final List<MoxplatformAndroidPlugin> _instances = new ArrayList<>();
   private BackgroundService service;
   private MethodChannel channel;
-  private Context context;
+  private static EventChannel notificationChannel;
+  public static EventSink notificationSink;
+
+    private Context context;
 
   public MoxplatformAndroidPlugin() {
     _instances.add(this);
@@ -58,6 +64,12 @@ public class MoxplatformAndroidPlugin extends BroadcastReceiver implements Flutt
     channel.setMethodCallHandler(this);
     context = flutterPluginBinding.getApplicationContext();
 
+    notificationChannel = new EventChannel(flutterPluginBinding.getBinaryMessenger(), "me.polynom/notification_stream");
+    notificationChannel.setStreamHandler(
+            this
+    );
+
+
     LocalBroadcastManager localBroadcastManager = LocalBroadcastManager.getInstance(this.context);
     localBroadcastManager.registerReceiver(this, new IntentFilter(methodChannelKey));
 
@@ -78,6 +90,18 @@ public class MoxplatformAndroidPlugin extends BroadcastReceiver implements Flutt
     Log.d(TAG, "Registered against registrar");
   }
 
+  @Override
+  public void onCancel(Object arguments) {
+    Log.d(TAG, "Removed listener");
+    notificationSink = null;
+  }
+
+  @Override
+  public void onListen(Object arguments, EventChannel.EventSink eventSink) {
+    Log.d(TAG, "Attached listener");
+    notificationSink = eventSink;
+  }
+
   /// Store the entrypoint handle and extra data for the background service.
   private void configure(long entrypointHandle, String extraData) {
     SharedPreferences prefs = context.getSharedPreferences(sharedPrefKey, Context.MODE_PRIVATE);
@@ -292,4 +316,9 @@ public class MoxplatformAndroidPlugin extends BroadcastReceiver implements Flutt
   public String getCacheDataPath() {
     return context.getCacheDir().getPath();
   }
+
+  @Override
+  public void eventStub(@NonNull NotificationEvent event) {
+    // Stub to trick pigeon into
+  }
 }

packages/moxplatform_android/android/src/main/java/me/polynom/moxplatform_android/NotificationReceiver.kt

@@ -7,22 +7,59 @@ import android.content.Intent
 import android.util.Log
 import androidx.core.app.NotificationManagerCompat
 import androidx.core.app.RemoteInput
+import me.polynom.moxplatform_android.Api.NotificationEvent
 
 class NotificationReceiver : BroadcastReceiver() {
-    override fun onReceive(context: Context, intent: Intent) {
+    private fun handleMarkAsRead(context: Context, intent: Intent) {
-        // If it is a mark as read, dismiss the entire notification and
+        Log.d("NotificationReceiver", "Marking ${intent.getStringExtra("jid")} as read")
-        // send a notification to the app.
+        val jidWrapper = intent.getStringExtra("jid") ?: ""
-        // TODO: Notify app
+        NotificationManagerCompat.from(context).cancel(intent.getLongExtra(MARK_AS_READ_ID_KEY, -1).toInt())
-        if (intent.action == MARK_AS_READ_ACTION) {
+        MoxplatformAndroidPlugin.notificationSink?.success(
-            Log.d("NotificationReceiver", "Marking ${intent.getStringExtra("jid")} as read")
+            NotificationEvent().apply {
-            NotificationManagerCompat.from(context).cancel(intent.getLongExtra(MARK_AS_READ_ID_KEY, -1).toInt())
+                // TODO: Use constant for key
-            return
+                // TODO: Fix
+                jid = jidWrapper
+                type = Api.NotificationEventType.MARK_AS_READ
+                payload = null
+            }.toList()
+        )
+
+        // Dismiss the notification
+        val notificationId = intent.getLongExtra("notification_id", -1).toInt()
+        if (notificationId != -1) {
+            NotificationManagerCompat.from(context).cancel(
+                notificationId,
+            )
+        } else {
+            Log.e("NotificationReceiver", "No id specified. Cannot dismiss notification")
         }
+    }
 
+    private fun handleReply(context: Context, intent: Intent) {
+        val jidWrapper = intent.getStringExtra("jid") ?: ""
         val remoteInput = RemoteInput.getResultsFromIntent(intent) ?: return
-
+        Log.d("NotificationReceiver", "Got a reply for ${jidWrapper}")
-        val title = remoteInput.getCharSequence(REPLY_TEXT_KEY).toString()
-        Log.d("NotificationReceiver", title)
         // TODO: Notify app
+        MoxplatformAndroidPlugin.notificationSink?.success(
+            NotificationEvent().apply {
+                // TODO: Use constant for key
+                jid = jidWrapper
+                type = Api.NotificationEventType.REPLY
+                payload = remoteInput.getCharSequence(REPLY_TEXT_KEY).toString()
+            }.toList()
+        )
+
+        // TODO: Update the notification to prevent showing the spinner
+    }
+
+    override fun onReceive(context: Context, intent: Intent) {
+        // TODO: We need to be careful to ensure that the Flutter engine is running.
+        //       If it's not, we have to start it. However, that's only an issue when we expect to
+        //       receive notifications while not running, i.e. Push Notifications.
+        when (intent.action) {
+            MARK_AS_READ_ACTION -> handleMarkAsRead(context, intent)
+            REPLY_ACTION -> handleReply(context, intent)
+            // TODO: Handle tap
+        }
     }
 }

packages/moxplatform_android/android/src/main/java/me/polynom/moxplatform_android/Notifications.kt

@@ -20,7 +20,11 @@ fun showMessagingNotification(context: Context, notification: Api.MessagingNotif
         // TODO: i18n
         setLabel("Reply")
     }.build()
-    val replyIntent = Intent(context, NotificationReceiver::class.java)
+    val replyIntent = Intent(context, NotificationReceiver::class.java).apply {
+        action = REPLY_ACTION
+        // TODO: Use a constant
+        putExtra("jid", notification.jid)
+    }
     val replyPendingIntent = PendingIntent.getBroadcast(
         context.applicationContext,
         0,
@@ -40,20 +44,29 @@ fun showMessagingNotification(context: Context, notification: Api.MessagingNotif
     // -> Mark as read action
     val markAsReadIntent = Intent(context, NotificationReceiver::class.java).apply {
         action = MARK_AS_READ_ACTION
-        // TODO: Put the JID here
+        // TODO: Use a constant
         putExtra("jid", notification.jid)
+        putExtra("notification_id", notification.id)
     }
     val markAsReadPendingIntent = PendingIntent.getBroadcast(
         context.applicationContext,
         0,
         markAsReadIntent,
-        0,
+        PendingIntent.FLAG_UPDATE_CURRENT,
     )
+    val markAsReadAction = NotificationCompat.Action.Builder(
+        // TODO: Wrong icon
+        R.drawable.ic_service_icon,
+        // TODO: i18n
+        "Mark as read",
+        markAsReadPendingIntent,
+    ).build()
 
     // -> Tap action
     // Thanks https://github.com/MaikuB/flutter_local_notifications/blob/master/flutter_local_notifications/android/src/main/java/com/dexterous/flutterlocalnotifications/FlutterLocalNotificationsPlugin.java#L246
     // TODO: Copy the interface of awesome_notifications
     val tapIntent = context.packageManager.getLaunchIntentForPackage(context.packageName)!!.apply {
+        // TODO: Use a constant
         putExtra("jid", notification.jid)
     }
     val tapPendingIntent = PendingIntent.getActivity(
@@ -115,13 +128,7 @@ fun showMessagingNotification(context: Context, notification: Api.MessagingNotif
 
         // Notification actions
         addAction(replyAction)
-        addAction(
+        addAction(markAsReadAction)
-            // TODO: Wrong icon
-            R.drawable.ic_service_icon,
-            // TODO: i18n
-            "Mark as read",
-            markAsReadPendingIntent,
-        )
     }.build()
 
     // Post the notification

packages/moxplatform_android/lib/src/notifications_android.dart

@@ -1,17 +1,32 @@
+import 'dart:async';
+import 'package:flutter/services.dart';
 import 'package:moxplatform_platform_interface/moxplatform_platform_interface.dart';
 
 class AndroidNotificationsImplementation extends NotificationsImplementation {
   final MoxplatformApi _api = MoxplatformApi();
 
+  final EventChannel _channel =
+      const EventChannel('me.polynom/notification_stream');
 
   @override
-  Future<void> createNotificationChannel(String title, String id, bool urgent) async {
+  Future<void> createNotificationChannel(
+    String title,
+    String id,
+    bool urgent,
+  ) async {
     return _api.createNotificationChannel(title, id, urgent);
   }
 
-
   @override
-  Future<void> showMessagingNotification(MessagingNotification notification) async {
+  Future<void> showMessagingNotification(
+    MessagingNotification notification,
+  ) async {
     return _api.showMessagingNotification(notification);
   }
+
+  @override
+  Stream<NotificationEvent> getEventStream() => _channel
+      .receiveBroadcastStream()
+      .cast<Object>()
+      .map(NotificationEvent.decode);
 }

packages/moxplatform_platform_interface/lib/src/api.g.dart

@@ -8,6 +8,12 @@ import 'dart:typed_data' show Float64List, Int32List, Int64List, Uint8List;
 import 'package:flutter/foundation.dart' show ReadBuffer, WriteBuffer;
 import 'package:flutter/services.dart';
 
+enum NotificationEventType {
+  markAsRead,
+  reply,
+  open,
+}
+
 class NotificationMessageContent {
   NotificationMessageContent({
     this.body,
@@ -133,6 +139,42 @@ class MessagingNotification {
   }
 }
 
+class NotificationEvent {
+  NotificationEvent({
+    required this.jid,
+    required this.type,
+    this.payload,
+  });
+
+  /// The JID the notification was for.
+  String jid;
+
+  /// The type of event.
+  NotificationEventType type;
+
+  /// An optional payload.
+  /// - type == NotificationType.reply: The reply message text.
+  /// Otherwise: undefined.
+  String? payload;
+
+  Object encode() {
+    return <Object?>[
+      jid,
+      type.index,
+      payload,
+    ];
+  }
+
+  static NotificationEvent decode(Object result) {
+    result as List<Object?>;
+    return NotificationEvent(
+      jid: result[0]! as String,
+      type: NotificationEventType.values[result[1]! as int],
+      payload: result[2] as String?,
+    );
+  }
+}
+
 class _MoxplatformApiCodec extends StandardMessageCodec {
   const _MoxplatformApiCodec();
   @override
@@ -140,12 +182,15 @@ class _MoxplatformApiCodec extends StandardMessageCodec {
     if (value is MessagingNotification) {
       buffer.putUint8(128);
       writeValue(buffer, value.encode());
-    } else if (value is NotificationMessage) {
+    } else if (value is NotificationEvent) {
       buffer.putUint8(129);
       writeValue(buffer, value.encode());
-    } else if (value is NotificationMessageContent) {
+    } else if (value is NotificationMessage) {
       buffer.putUint8(130);
       writeValue(buffer, value.encode());
+    } else if (value is NotificationMessageContent) {
+      buffer.putUint8(131);
+      writeValue(buffer, value.encode());
     } else {
       super.writeValue(buffer, value);
     }
@@ -157,8 +202,10 @@ class _MoxplatformApiCodec extends StandardMessageCodec {
       case 128: 
         return MessagingNotification.decode(readValue(buffer)!);
       case 129: 
-        return NotificationMessage.decode(readValue(buffer)!);
+        return NotificationEvent.decode(readValue(buffer)!);
       case 130: 
+        return NotificationMessage.decode(readValue(buffer)!);
+      case 131: 
         return NotificationMessageContent.decode(readValue(buffer)!);
       default:
         return super.readValueOfType(type, buffer);
@@ -273,4 +320,26 @@ class MoxplatformApi {
       return (replyList[0] as String?)!;
     }
   }
+
+  Future<void> eventStub(NotificationEvent arg_event) async {
+    final BasicMessageChannel<Object?> channel = BasicMessageChannel<Object?>(
+        'dev.flutter.pigeon.moxplatform_platform_interface.MoxplatformApi.eventStub', codec,
+        binaryMessenger: _binaryMessenger);
+    final List<Object?>? replyList =
+        await channel.send(<Object?>[arg_event]) as List<Object?>?;
+    if (replyList == null) {
+      throw PlatformException(
+        code: 'channel-error',
+        message: 'Unable to establish connection on channel.',
+      );
+    } else if (replyList.length > 1) {
+      throw PlatformException(
+        code: replyList[0]! as String,
+        message: replyList[1] as String?,
+        details: replyList[2],
+      );
+    } else {
+      return;
+    }
+  }
 }

packages/moxplatform_platform_interface/lib/src/notifications.dart

@@ -1,7 +1,10 @@
+import 'dart:async';
 import 'package:moxplatform_platform_interface/src/api.g.dart';
 
 abstract class NotificationsImplementation {
   Future<void> createNotificationChannel(String title, String id, bool urgent);
 
   Future<void> showMessagingNotification(MessagingNotification notification);
+
+  Stream<NotificationEvent> getEventStream();
 }

packages/moxplatform_platform_interface/lib/src/notifications_stub.dart

@@ -1,3 +1,4 @@
+import 'dart:async';
 import 'package:moxplatform_platform_interface/src/api.g.dart';
 import 'package:moxplatform_platform_interface/src/notifications.dart';
 
@@ -7,4 +8,9 @@ class StubNotificationsImplementation extends NotificationsImplementation {
 
   @override
   Future<void> showMessagingNotification(MessagingNotification notification) async {}
+
+  @override
+  Stream<NotificationEvent> getEventStream() {
+    return StreamController<NotificationEvent>().stream;
+  }
 }

pigeons/api.dart

@@ -72,6 +72,31 @@ class MessagingNotification {
   final List<NotificationMessage?> messages;
 }
 
+enum NotificationEventType {
+  markAsRead,
+  reply,
+  open,
+}
+
+class NotificationEvent {
+  const NotificationEvent(
+    this.jid,
+    this.type,
+    this.payload,
+  );
+
+  /// The JID the notification was for.
+  final String jid;
+
+  /// The type of event.
+  final NotificationEventType type;
+
+  /// An optional payload.
+  /// - type == NotificationType.reply: The reply message text.
+  /// Otherwise: undefined.
+  final String? payload;
+}
+
 @HostApi()
 abstract class MoxplatformApi {
   void createNotificationChannel(String title, String id, bool urgent);
@@ -81,4 +106,6 @@ abstract class MoxplatformApi {
   String getPersistentDataPath();
 
   String getCacheDataPath();
+
+  void eventStub(NotificationEvent event);
 }