Erweiterungen

lib/tools/message_service.dart

@@ -0,0 +1,92 @@
+typedef EventHandler<T> = void Function(T args);
+
+/// The Custom Events Service that can publish data
+/// from one module/component to another.
+class MessageService {
+  static final MessageService _singleton = new MessageService._internal();
+
+  Map<String, dynamic> _callbackMaps = Map<String, dynamic>();
+
+  factory MessageService() {
+    return _singleton;
+  }
+
+  MessageService._internal();
+
+  /// Subscribe to an event topic. Events that
+  /// get posted to that topic will trigger
+  /// the provided handler
+  /// * [topic] The topic to subscribe to
+  /// * [handler] The event handler
+  subscribe<T>(String topic, EventHandler<T>? handler) {
+    List<EventHandler<T>>? handlers = _callbackMaps[topic];
+    if (handler == null) {
+      _callbackMaps.remove(topic);
+      return;
+    }
+    if (handlers == null) {
+      handlers = [];
+      _callbackMaps[topic] = handlers;
+    }
+    handlers.add(handler);
+  }
+
+  /// Unsubscribe from the given topic. Your handler
+  /// will no longer recieve events published to this
+  /// topic
+  /// * [topic] The topic to unsubscribe from
+  /// * [handler] The event handler
+  unsubscribe<T>(String topic, {EventHandler<T>? handler}) {
+    if (handler == null) {
+      _callbackMaps.remove(topic);
+      return;
+    }
+
+    final handlers = _callbackMaps[topic];
+    if (handlers == null) {
+      return;
+    }
+
+    // We now need to remove specific handler
+    final handlerIndex = handlers.indexOf(handler);
+
+    if (handlerIndex >= 0) {
+      handlers.removeAt(handlerIndex);
+      if (handlers.length == 0) {
+        _callbackMaps.remove(topic);
+      }
+    }
+  }
+
+  /// Publish an Event to given topic.
+  /// * [topic] The topic to publish to
+  /// * [arguments] The arguments to pass to
+  /// the handlers. It's better if custom
+  /// Structure is used.
+  publish<T>(String topic, {T? arguments}) {
+    List<EventHandler<T>>? handlers;
+    try {
+      handlers = _callbackMaps[topic];
+    } catch (_) {
+      throw FormatException(
+          'Type mismatch between Publisher Arguments and what Subscriber Expects');
+    }
+
+    if (handlers == null || handlers.isEmpty) {
+      return;
+    }
+
+    handlers.forEach((EventHandler<T> handler) {
+      try {
+        if (arguments != null) {
+          handler(arguments);
+        } else {
+          throw 'arguments null';
+        }
+      } catch (_) {
+        print('Error Occured while executing handler for topic $topic');
+        print(_);
+      }
+    });
+  }
+}

lib/tools/mqtt_core.dart

@@ -2,7 +2,6 @@ import 'dart:async';
 import 'dart:convert';
 import 'dart:math';
 
-import 'package:flutter_broadcast_receiver/flutter_broadcast_receiver.dart';
 import 'package:isar/isar.dart';
 import 'package:mqtt_client/mqtt_client.dart';
 import 'package:mqtt_client/mqtt_server_client.dart';
@@ -16,6 +15,7 @@ import 'package:stadtbote/protocoll_messages/payload_container.dart';
 import 'package:stadtbote/protocoll_messages/start_session_message.dart';
 import 'package:stadtbote/protocoll_messages/sync_jobs_message.dart';
 import 'package:stadtbote/tools/connection_status.dart';
+import 'package:stadtbote/tools/message_service.dart';
 import 'package:stadtbote/tools/messages/job_deleted_message.dart';
 import 'package:stadtbote/tools/messages/job_received_message.dart';
 import 'package:stadtbote/tools/messages/message_message.dart';
@@ -90,7 +90,7 @@ class MqttCore {
       } else {
         Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!');
 
-        BroadcastReceiver().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+        MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
 
         _startReconnectTimer();
       }
@@ -100,7 +100,7 @@ class MqttCore {
 
       Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!: $ex');
 
-      BroadcastReceiver().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+      MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
     }
   }
 
@@ -194,7 +194,7 @@ class MqttCore {
     _mqttClient.subscribe(_fromServerTopic, MqttQos.exactlyOnce);
     Util.logInfo('subscribed to topic: $_fromServerTopic');
 
-    BroadcastReceiver().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
+    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
   }
 
   void _sessionMessageReceived(payloadData) async {
@@ -208,7 +208,7 @@ class MqttCore {
       await MyApp.isarInstance.status.put(statusMessage);
     });
 
-    BroadcastReceiver().publish<StatusInfoUpdate>(Constants.messageTopicStatusInfoUpdate, arguments: StatusInfoUpdate(status: statusMessage));
+    MessageService().publish<StatusInfoUpdate>(Constants.messageTopicStatusInfoUpdate, arguments: StatusInfoUpdate(status: statusMessage));
   }
 
   void _rankingMessageReceived(payloadData) async {
@@ -231,7 +231,7 @@ class MqttCore {
         MyApp.isarInstance.jobs.putSync(newJob);
       });
 
-      BroadcastReceiver().publish<JobReceivedMessage>(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob));
+      MessageService().publish<JobReceivedMessage>(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob));
     }
   }
 
@@ -245,7 +245,7 @@ class MqttCore {
         await MyApp.isarInstance.jobs.delete(jobId);
       });
 
-      BroadcastReceiver().publish<JobDeletedMessage>(Constants.messageTopicJobDeleted, arguments: JobDeletedMessage(jobId: jobId));
+      MessageService().publish<JobDeletedMessage>(Constants.messageTopicJobDeleted, arguments: JobDeletedMessage(jobId: jobId));
     }
   }
 
@@ -270,7 +270,7 @@ class MqttCore {
       await MyApp.isarInstance.messages.put(newMessage);
     });
 
-    BroadcastReceiver().publish<MessageReceivedMessage>(Constants.messageTopicMessageReceived, arguments: MessageReceivedMessage(message: newMessage));
+    MessageService().publish<MessageReceivedMessage>(Constants.messageTopicMessageReceived, arguments: MessageReceivedMessage(message: newMessage));
   }
 
   void _startReconnectTimer() {
@@ -290,7 +290,7 @@ class MqttCore {
 
     Util.logInfo("_clientDisconnected called");
 
-    BroadcastReceiver().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
 
     _startReconnectTimer();
   }