Assecutor/stadtbote-app
4
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into new_appserver

Browse Source 
# Conflicts:
#	lib/content_views/approach_area_page.dart
#	lib/content_views/assigned_jobs_page.dart
#	lib/content_views/main_page.dart
#	lib/content_views/task_list_page.dart
#	lib/main.dart
#	lib/popups/accept_jobs_popup.dart
#	lib/protocoll_messages/logout_message.dart
#	lib/protocoll_messages/logout_message.g.dart
#	lib/tools/constants.dart
#	lib/tools/mqtt_core.dart
#	pubspec.lock
#	pubspec.yaml
This commit is contained in:
Sven committed 2024-03-26 12:56:04 +01:00
2 parents 12a351e604 a683ea0ff1
commit 880cdd65ff
74 files changed
+2463 -441
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

lib/tools/mqtt_core.dart
+172 -92
View File
@@ -1,19 +1,25 @@
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'dart:math';
import 'package:archive/archive.dart';
import 'package:device_info_plus/device_info_plus.dart';
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/enqueued_message.dart';
import 'package:stadtbote/model/log_entry.dart';
import 'package:stadtbote/model/ranking.dart';
import 'package:stadtbote/model/status.dart';
import 'package:stadtbote/protocoll_messages/acknowledge_job.dart';
import 'package:stadtbote/protocoll_messages/bootstrap_request.dart';
import 'package:stadtbote/protocoll_messages/bootstrap_response_payload.dart';
import 'package:stadtbote/protocoll_messages/device_log_request.dart';
import 'package:stadtbote/protocoll_messages/device_log_response.dart';
import 'package:stadtbote/protocoll_messages/device_object.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/payload_container.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';
@@ -28,6 +34,7 @@ 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 'package:uuid/uuid.dart';
import '../main.dart';
import '../model/session_status.dart';
@@ -49,7 +56,6 @@ class MqttCore {
late String _brokerUri;
late int _brokerPort;
late String _fromServerTopic;
late String _clientId;
late String _toServerTopic;
late MqttServerClient _mqttClient;
@@ -94,7 +100,6 @@ class MqttCore {
_mqttClient.keepAlivePeriod = 20;
_mqttClient.autoReconnect = false;
_mqttClient.onDisconnected = _clientDisconnected;
_mqttClient.connectionMessage = MqttConnectMessage()
.withClientIdentifier(await Util.getAppUuid())
.withWillQos(MqttQos.atMostOnce)
@@ -129,73 +134,96 @@ class MqttCore {
}
}
publishText(String subtopic, String text) {
publishBinary(String topic, List<int> data) {
var buffer = Uint8Buffer();
buffer.addAll(data);
_mqttClient.publishMessage(topic, MqttQos.exactlyOnce, buffer);
Util.logInfo("binary message send to $topic");
}
publishText(String topic, String text) {
var buffer = Uint8Buffer();
buffer.addAll(utf8.encode(text));
_mqttClient.publishMessage(topic, MqttQos.exactlyOnce, buffer);
Util.logInfo("message\n$text\nsend to $topic");
}
publishPayloadMessage(String topic, Map<String, dynamic> payload) {
var payloadContainer = PayloadContainer();
payloadContainer.id = const Uuid().v4().toString();
payloadContainer.recv = topic;
payloadContainer.store = true;
payloadContainer.payload = payload;
var json = jsonEncode(payloadContainer.toJson());
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");
publishText(_toServerTopic, json);
} catch (e) {
Util.logException(e.toString());
_enqueueMessage(_toServerTopic, jsonEncode(payloadContainer.toJson()));
}
}
void _setupSubscriptions() async {
Util.logInfo('_setupSubscriptions called');
_mqttClient.updates?.listen((List<MqttReceivedMessage<MqttMessage>> messages) {
_mqttClient.updates?.listen((List<MqttReceivedMessage<MqttMessage>> c) {
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;
final MqttPublishMessage message = c[0].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');
Util.logInfo('message received\ntopic: $topic\nmessage: $payload');
var topicArray = topic.split('/');
var messageType = topicArray[3];
Map<String, dynamic> messageData = jsonDecode(payload);
switch(messageType) {
case 'session':
_sessionMessageReceived(jsonDecode(payload));
break;
for (var entry in messageData.keys) {
if (entry == 'payload') {
var payload = messageData[entry];
for (var payloadEntry in payload.keys) {
var payloadData = payload[payloadEntry];
case 'status':
_statusMessageReceived(jsonDecode(payload));
break;
switch(payloadEntry) {
case 'session':
_sessionMessageReceived(payloadData);
break;
case 'ranking':
_rankingMessageReceived(jsonDecode(payload));
break;
case 'status':
_statusMessageReceived(payloadData);
break;
case 'job':
_jobMessageReceived(jsonDecode(payload));
break;
case 'ranks':
_rankingMessageReceived(payloadData);
break;
case 'job_del':
_jobDeleteReceived(jsonDecode(payload));
break;
case 'jobs':
_jobMessageReceived(payloadData);
break;
case 'sync_request':
_syncRequestReceived(jsonDecode(payload));
break;
case 'job_del':
_jobDeleteReceived(payloadData);
break;
case "mssg":
_messageMessageReceived(payload);
break;
case 'sync_request':
_syncRequestReceived(payloadData);
break;
default:
Util.logInfo('Unknown topic: $messageType');
break;
case "mssg":
_messageMessageReceived(payloadData);
break;
case "upload_device_log":
_messageUploadDeviceLogReceived(payloadData);
break;
default:
Util.logInfo('Unknown topic: $payloadEntry');
break;
}
}
}
}
@@ -204,9 +232,8 @@ class MqttCore {
}
});
var topic = "$_fromServerTopic#";
_mqttClient.subscribe(topic, MqttQos.exactlyOnce);
Util.logInfo('subscribed to topic: $topic');
_mqttClient.subscribe(_fromServerTopic, MqttQos.exactlyOnce);
Util.logInfo('subscribed to topic: $_fromServerTopic');
MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
}
@@ -216,7 +243,9 @@ class MqttCore {
await _sendSyncMessage();
var sessionStatusMessage = SessionStatus.fromJson(payloadData);
await _sendEnqueuedMessages();
var sessionStatusMessage = SessionStatus.fromJson(payloadData['status']);
await MyApp.isarInstance.writeTxn(() async {
await MyApp.isarInstance.sessionStatus.clear();
@@ -240,42 +269,52 @@ class MqttCore {
}
void _rankingMessageReceived(List<dynamic> payloadData) async {
for (var entry in payloadData) {
var rank = Rank.fromJson(entry);
await MyApp.isarInstance.writeTxn(() async {
await MyApp.isarInstance.ranks.put(rank);
});
}
Map<String, dynamic> rankingMap = {};
rankingMap["ranks"] = payloadData;
var ranking = Ranking.fromJson(rankingMap);
await MyApp.isarInstance.writeTxn(() async {
await MyApp.isarInstance.rankings.put(ranking);
});
MessageService().publish<RankingUpdated>(Constants.messageTopicRankingUpdated, arguments: RankingUpdated());
}
void _jobMessageReceived(payloadData) async {
Job newJob = Job.fromJson(payloadData);
for (var foundJobData in payloadData) {
Job newJob = Job.fromJson(foundJobData);
MyApp.isarInstance.writeTxnSync(() async {
MyApp.isarInstance.jobs.putSync(newJob);
});
await MyApp.isarInstance.writeTxn(() async {
await MyApp.isarInstance.jobs.put(newJob);
});
if (newJob.state == Constants.stateAssigned) {
if (newJob.global) {
await Util.playMp3('aaa.mp3');
} else {
await Util.playMp3('job_received.mp3');
if (newJob.state == Constants.stateAssigned) {
if (newJob.global) {
await Util.playMp3('aaa.mp3');
} else {
await Util.playMp3('job_received.mp3');
}
}
}
MessageService().publish<JobReceivedMessage>(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob));
try {
MessageService().publish<JobReceivedMessage>(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob));
} catch (ex) {
Util.logException("publish messageTopicJobReceived failed!");
}
_acknowlegdeJob(newJob);
}
}
void _jobDeleteReceived(payloadData) async {
var jobDelMessage = JobDelMessage.fromJson(payloadData);
void _jobDeleteReceived(List<dynamic> payloadData) async {
Map<String, dynamic> map = {};
map["jobDel"] = payloadData;
var jobDelMessage = JobDelMessage.fromJson(map);
await MyApp.isarInstance.writeTxn(() async {
await MyApp.isarInstance.jobs.delete(jobDelMessage.id);
});
for(var jobId in jobDelMessage.jobDel) {
await MyApp.isarInstance.writeTxn(() async {
await MyApp.isarInstance.jobs.delete(jobId);
});
MessageService().publish<JobDeletedMessage>(Constants.messageTopicJobDeleted, arguments: JobDeletedMessage(id: jobDelMessage.id));
}
@@ -334,19 +373,22 @@ class MqttCore {
void _processBootstrap() async {
try {
String? responseText = "";
var deviceInfoPlugin = DeviceInfoPlugin();
AndroidDeviceInfo androidInfo = await deviceInfoPlugin.androidInfo;
BaseDeviceInfo deviceInfo = await deviceInfoPlugin.deviceInfo;
var deviceObject = DeviceObject();
deviceObject.manufacturer = "sven";
deviceObject.model = "emulator";
deviceObject.os = "hat es";
deviceObject.osVer = "0.0";
deviceObject.manufacturer = deviceInfo.data['manufacturer'];
deviceObject.model = deviceInfo.data['model'];
deviceObject.os = androidInfo.version.release;
deviceObject.osVer = androidInfo.version.sdkInt.toString();
var handshakeMessage = BootstrapRequest();
handshakeMessage.app = "sb";
handshakeMessage.mqttClientId = await Util.getAppUuid();
handshakeMessage.deviceObject = deviceObject;
handshakeMessage.ver = 200;
handshakeMessage.ver = await Util.getAppVersion();
//var jsonMap = handshakeMessage.toJson();
//var json = jsonEncode(jsonMap);
@@ -405,12 +447,12 @@ class MqttCore {
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();
trackMessage.pos!.time = position.timestamp.toIso8601String();
trackMessage.pos!.lat = position.latitude;
trackMessage.pos!.long = position.longitude;
trackMessage.pos!.acc = position.accuracy.toInt();
publishText("track", jsonEncode(answer.toJson()));
publishPayloadMessage("track", trackMessage.toJson());
}
});
}
@@ -421,4 +463,42 @@ class MqttCore {
_trackTimer = null;
}
}
void _enqueueMessage(String receiver, String payload) async {
var enqueuedMessage = EnqueuedMessage();
enqueuedMessage.topic = receiver;
enqueuedMessage.payload = payload;
await MyApp.isarInstance.writeTxn(() async {
await MyApp.isarInstance.enqueuedMessages.put(enqueuedMessage);
});
}
_sendEnqueuedMessages() async {
var enqueuedMessages = await MyApp.isarInstance.enqueuedMessages.where().findAll();
if (enqueuedMessages.isNotEmpty) {
for (var enqueuedMessage in enqueuedMessages) {
try {
publishText(enqueuedMessage.topic, enqueuedMessage.payload);
await MyApp.isarInstance.writeTxn(() async {
await MyApp.isarInstance.enqueuedMessages.delete(enqueuedMessage.id);
});
} catch (ex) {
Util.logInfo("resend enqueued message failed: $ex");
}
}
}
}
void _acknowlegdeJob(Job job) {
var message = AcknowledgeJob();
message.received = AcknowledgeJobReceived();
message.received.id = job.id;
message.received.state = job.state;
message.received.ver = job.ver;
message.received.utc = DateTime.now().millisecondsSinceEpoch;
MqttCore.instance.publishPayloadMessage('job', message.toJson());
}
}