import 'dart:async'; import 'dart:convert'; import 'dart:io'; import 'dart:math'; import 'package:isar/isar.dart'; import 'package:mqtt_client/mqtt_client.dart'; import 'package:mqtt_client/mqtt_server_client.dart'; import 'package:stadtbote/model/rank.dart'; import 'package:stadtbote/model/status.dart'; import 'package:stadtbote/protocoll_messages/bootstrap_request.dart'; import 'package:stadtbote/protocoll_messages/bootstrap_response_payload.dart'; import 'package:stadtbote/model/job.dart'; import 'package:stadtbote/model/message.dart'; import 'package:stadtbote/protocoll_messages/device_object.dart'; import 'package:stadtbote/protocoll_messages/job_del_message.dart'; import 'package:stadtbote/protocoll_messages/start_session_message.dart'; import 'package:stadtbote/protocoll_messages/sync_jobs_message.dart'; import 'package:stadtbote/protocoll_messages/track_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'; import 'package:stadtbote/tools/messages/ranking_updated.dart'; import 'package:stadtbote/tools/messages/session_info_update.dart'; import 'package:stadtbote/tools/messages/status_info_update.dart'; // ignore: depend_on_referenced_packages import 'package:typed_data/typed_buffers.dart'; import 'package:http/http.dart' as http; import '../main.dart'; import '../model/session_status.dart'; import 'constants.dart'; import 'util.dart'; enum ConnectionStateEnum { notConnected, connected } class MqttCore { static late MqttCore instance; late ConnectionStateEnum connectionState = ConnectionStateEnum.notConnected; late bool _disconnectInProgress = false; late String _brokerUri; late int _brokerPort; late String _fromServerTopic; late String _clientId; late String _toServerTopic; late MqttServerClient _mqttClient; late int _reconnectDelay = 0; Timer? _reconnectTimer; Timer? _trackTimer; MqttCore() { _disconnectInProgress = false; _processBootstrap(); } static setup() { instance = MqttCore(); } diconnect() { _disconnectInProgress = true; _cancelReconnectTimer(); _cancelTrackTimer(); _mqttClient.disconnect(); connectionState = ConnectionStateEnum.notConnected; } _connectToBroker() async { try { String clientId = await Util.getAppUuid(); Util.logInfo("client id: $clientId"); Util.logInfo("broker uri: $_brokerUri"); Util.logInfo("broker port: $_brokerPort"); _mqttClient = MqttServerClient.withPort(_brokerUri, clientId, _brokerPort); _mqttClient.logging(on: false); _mqttClient.keepAlivePeriod = 20; _mqttClient.autoReconnect = false; _mqttClient.onDisconnected = _clientDisconnected; _mqttClient.connectionMessage = MqttConnectMessage() .withClientIdentifier(await Util.getAppUuid()) .withWillQos(MqttQos.atMostOnce) .withWillTopic('willTopic') .withWillMessage('willMessage') .authenticateAs("app", "apppwd"); await _mqttClient.connect(); if (_mqttClient.connectionStatus?.state == MqttConnectionState.connected) { Util.logInfo('Verbunden mit dem MQTT-Broker!'); _reconnectDelay = 0; _setupSubscriptions(); _initSession(); } else { Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!'); MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); _startReconnectTimer(); } } catch (ex) { _startReconnectTimer(); Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!: $ex'); MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); } } publishText(String subtopic, String text) { try { var buffer = Uint8Buffer(); buffer.addAll(utf8.encode(text)); _mqttClient.publishMessage(subtopic, MqttQos.exactlyOnce, buffer); var topic = "$_toServerTopic$subtopic/$_clientId"; _mqttClient.publishMessage( topic, MqttQos.exactlyOnce, buffer); Util.logInfo("send message $text to $topic"); } catch (e) { Util.logException(e.toString()); } } void _setupSubscriptions() async { Util.logInfo('_setupSubscriptions called'); _mqttClient.updates?.listen((List> messages) { try { for (var mqttMessage in messages) { final MqttPublishMessage message = mqttMessage.payload as MqttPublishMessage; String payload = utf8.decode(MqttPublishPayload.bytesToStringAsString(message.payload.message).codeUnits); var topic = message.variableHeader?.topicName; if (topic!.endsWith(_clientId)) { Util.logInfo('message received'); Util.logInfo('topic: $topic'); Util.logInfo('mesage: $payload'); var topicArray = topic.split('/'); var messageType = topicArray[3]; switch(messageType) { case 'session': _sessionMessageReceived(jsonDecode(payload)); break; case 'status': _statusMessageReceived(jsonDecode(payload)); break; case 'ranking': _rankingMessageReceived(jsonDecode(payload)); break; case 'job': _jobMessageReceived(jsonDecode(payload)); break; case 'job_del': _jobDeleteReceived(jsonDecode(payload)); break; case 'sync_request': _syncRequestReceived(jsonDecode(payload)); break; case "mssg": _messageMessageReceived(payload); break; default: Util.logInfo('Unknown topic: $messageType'); break; } } } } catch(ex) { Util.logException(ex.toString()); } }); var topic = "$_fromServerTopic#"; _mqttClient.subscribe(topic, MqttQos.exactlyOnce); Util.logInfo('subscribed to topic: $topic'); MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected)); } void _sessionMessageReceived(payloadData) async { connectionState = ConnectionStateEnum.connected; await _sendSyncMessage(); var sessionStatusMessage = SessionStatus.fromJson(payloadData); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.sessionStatus.clear(); await MyApp.isarInstance.sessionStatus.put(sessionStatusMessage); }); MessageService().publish(Constants.messageTopicSessionStatusInfoUpdate, arguments: SessionStatusInfoUpdate(status: sessionStatusMessage)); _startTrackTimer(); } void _statusMessageReceived(payloadData) async { var statusMessage = Status.fromJson(payloadData); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.status.clear(); await MyApp.isarInstance.status.put(statusMessage); }); MessageService().publish(Constants.messageTopicStatusInfoUpdated, arguments: StatusInfoUpdate(status: statusMessage)); } void _rankingMessageReceived(List payloadData) async { for (var entry in payloadData) { var rank = Rank.fromJson(entry); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.ranks.put(rank); }); } MessageService().publish(Constants.messageTopicRankingUpdated, arguments: RankingUpdated()); } void _jobMessageReceived(payloadData) async { Job newJob = Job.fromJson(payloadData); MyApp.isarInstance.writeTxnSync(() async { MyApp.isarInstance.jobs.putSync(newJob); }); if (newJob.state == Constants.stateAssigned) { if (newJob.global) { await Util.playMp3('aaa.mp3'); } else { await Util.playMp3('job_received.mp3'); } } MessageService().publish(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob)); } void _jobDeleteReceived(payloadData) async { var jobDelMessage = JobDelMessage.fromJson(payloadData); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.jobs.delete(jobDelMessage.id); }); MessageService().publish(Constants.messageTopicJobDeleted, arguments: JobDeletedMessage(id: jobDelMessage.id)); } void _syncRequestReceived(payloadData) async { Util.logInfo('_syncRequestReceived'); _sendSyncMessage(); } void _messageMessageReceived(payloadData) async { var newMessage = await Message.mergeFromJson(payloadData); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.messages.put(newMessage); }); await Util.playMp3('asbeep.mp3'); MessageService().publish(Constants.messageTopicMessageReceived, arguments: MessageReceivedMessage(message: newMessage)); } void _startReconnectTimer() { connectionState = ConnectionStateEnum.notConnected; if (!_disconnectInProgress) { Util.logInfo("_startReconnectTimer called"); _reconnectDelay = min(_reconnectDelay + 1, 30); _reconnectTimer = Timer(Duration(seconds: _reconnectDelay), () { _processBootstrap(); }); } } void _cancelReconnectTimer() { if (_reconnectTimer != null && _reconnectTimer!.isActive) { _reconnectTimer!.cancel(); } } void _clientDisconnected() { connectionState = ConnectionStateEnum.notConnected; _cancelTrackTimer(); Util.logInfo("_clientDisconnected called"); MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); if (!_disconnectInProgress) { _startReconnectTimer(); } } void _processBootstrap() async { try { String? responseText = ""; var deviceObject = DeviceObject(); deviceObject.manufacturer = "sven"; deviceObject.model = "emulator"; deviceObject.os = "hat es"; deviceObject.osVer = "0.0"; var handshakeMessage = BootstrapRequest(); handshakeMessage.app = "sb"; handshakeMessage.mqttClientId = await Util.getAppUuid(); handshakeMessage.deviceObject = deviceObject; handshakeMessage.ver = 200; //var jsonMap = handshakeMessage.toJson(); //var json = jsonEncode(jsonMap); var uri = "http://test-app-server.dns.army:8080/bootstrap"; Util.logInfo("Bootstrap request: $uri"); var response = await http.get(Uri.parse(uri)); responseText = response.body; Util.logInfo("Bootstrap response: $responseText"); var bootstrapResponse = BootstrapResponse.fromJson(jsonDecode(responseText)); var brokerUriAndPortArray = bootstrapResponse.broker.replaceFirst("mqtt://", "").split(":"); _brokerUri = brokerUriAndPortArray[0]; _brokerPort = int.parse(brokerUriAndPortArray[1]); _clientId = bootstrapResponse.id; _fromServerTopic = bootstrapResponse.fromServer; _toServerTopic = bootstrapResponse.toServer; _connectToBroker(); } catch (ex) { _startReconnectTimer(); } } _initSession() { var startSessionMessage = StartSessionMessage(); var jsonMap = startSessionMessage.toJson(); var json = jsonEncode(jsonMap); publishText("session", json); } _sendSyncMessage() async { var jobs = await MyApp.isarInstance.jobs.filter().stateGreaterThan(-1).findAll(); var answer = SyncJobsMessage(); answer.utcUnixTime = DateTime.now().toUtc().millisecondsSinceEpoch ~/ 1000; answer.sync = SyncJobMessageSync(); answer.sync.jobs = []; for (var job in jobs) { SyncJobMessageJob syncJob = SyncJobMessageJob(); syncJob.state = job.state; syncJob.id = job.id; syncJob.ver = job.ver.toString(); answer.sync.jobs.add(syncJob); } MqttCore.instance.publishText("sync", jsonEncode(answer.toJson())); } void _startTrackTimer() { _trackTimer ??= Timer.periodic(const Duration(seconds: 60), (Timer t) { var answer = MyApp.currentPosition; if (answer != null) { var trackMessage = TrackMessage(); trackMessage.pos = TrackPosition(); trackMessage.pos.time = answer.timestamp.toIso8601String(); trackMessage.pos.lat = answer.latitude; trackMessage.pos.long = answer.longitude; trackMessage.pos.acc = answer.accuracy.toInt(); publishText("track", jsonEncode(answer.toJson())); } }); } void _cancelTrackTimer() { if (_trackTimer != null && _trackTimer!.isActive) { _trackTimer!.cancel(); _trackTimer = null; } } }