723 lines
23 KiB
Dart
723 lines
23 KiB
Dart
import 'dart:async';
|
|
import 'dart:convert';
|
|
import 'dart:math';
|
|
|
|
import 'package:archive/archive.dart';
|
|
import 'package:dart_mq/dart_mq.dart';
|
|
import 'package:device_info_plus/device_info_plus.dart';
|
|
import 'package:http/http.dart' as http;
|
|
import 'package:isar/isar.dart';
|
|
import 'package:mqtt5_client/mqtt5_client.dart';
|
|
import 'package:mqtt5_client/mqtt5_server_client.dart';
|
|
import 'package:stadtbote/model/enqueued_message.dart';
|
|
import 'package:stadtbote/model/job.dart';
|
|
import 'package:stadtbote/model/log_entry.dart';
|
|
import 'package:stadtbote/model/chat_message.dart';
|
|
import 'package:stadtbote/model/location_status.dart';
|
|
import 'package:stadtbote/model/rank.dart';
|
|
import 'package:stadtbote/model/status.dart';
|
|
import 'package:stadtbote/network_messages/acknowledge_job.dart';
|
|
import 'package:stadtbote/network_messages/acknowledge_message.dart';
|
|
import 'package:stadtbote/network_messages/bootstrap_request.dart';
|
|
import 'package:stadtbote/network_messages/bootstrap_response_payload.dart';
|
|
import 'package:stadtbote/network_messages/connection_lost_message.dart';
|
|
import 'package:stadtbote/network_messages/device_log_request.dart';
|
|
import 'package:stadtbote/network_messages/device_log_response.dart';
|
|
import 'package:stadtbote/network_messages/device_object.dart';
|
|
import 'package:stadtbote/network_messages/job_del_message.dart';
|
|
import 'package:stadtbote/network_messages/start_session_message.dart';
|
|
import 'package:stadtbote/network_messages/status_info_update.dart';
|
|
import 'package:stadtbote/network_messages/sync_jobs_message.dart';
|
|
import 'package:stadtbote/network_messages/track_message.dart';
|
|
import 'package:stadtbote/tools/connection_status.dart';
|
|
import 'package:stadtbote/tools/message_header.dart';
|
|
import 'package:stadtbote/tools/messages/publish_payload_message.dart';
|
|
import 'package:stadtbote/tools/messages/current_connection_status_message.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';
|
|
|
|
// ignore: unused_import
|
|
import 'package:uuid/uuid.dart';
|
|
|
|
import '../main.dart';
|
|
import '../model/session_status.dart';
|
|
import 'constants.dart';
|
|
import 'util.dart';
|
|
|
|
enum ConnectionStateEnum { notConnected, connected }
|
|
|
|
class MqttCore with ProducerMixin, ConsumerMixin {
|
|
static late MqttCore instance;
|
|
|
|
late ConnectionStateEnum connectionState = ConnectionStateEnum.notConnected;
|
|
|
|
late bool _disconnectInProgress = false;
|
|
|
|
late String _brokerUri;
|
|
late String _clientId;
|
|
late int _brokerPort;
|
|
late String _fromServerTopic;
|
|
late String _toServerTopic;
|
|
|
|
MqttServerClient? _mqttClient;
|
|
|
|
late int _reconnectDelay = 0;
|
|
|
|
Timer? _reconnectTimer;
|
|
|
|
Timer? _trackTimer;
|
|
|
|
MqttCore() {
|
|
instance = this;
|
|
|
|
initMessages();
|
|
|
|
_disconnectInProgress = false;
|
|
|
|
_processBootstrap();
|
|
}
|
|
|
|
ConnectionStateEnum getConnectionStatus() {
|
|
return connectionState;
|
|
}
|
|
|
|
_disconnectMqttClient() async {
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicConnectionStatusChanged,
|
|
payload: ConnectionStatus(status: ConnectionStatusEnum.notConnected),
|
|
);
|
|
|
|
_disconnectInProgress = true;
|
|
|
|
_cancelReconnectTimer();
|
|
|
|
_cancelTrackTimer();
|
|
|
|
if (_mqttClient != null) {
|
|
_mqttClient!.disconnect();
|
|
|
|
_mqttClient = null;
|
|
}
|
|
|
|
connectionState = ConnectionStateEnum.notConnected;
|
|
|
|
_disconnectInProgress = false;
|
|
}
|
|
|
|
_connectToBroker() async {
|
|
try {
|
|
String clientId = Util.getAppUuid();
|
|
Util.logInfo("client id: $clientId");
|
|
Util.logInfo("broker uri: $_brokerUri");
|
|
Util.logInfo("broker port: $_brokerPort");
|
|
|
|
if (_mqttClient == null) {
|
|
_mqttClient = MqttServerClient.withPort(_brokerUri, clientId, _brokerPort);
|
|
_mqttClient!.logging(on: false);
|
|
_mqttClient!.keepAlivePeriod = 60;
|
|
_mqttClient!.disconnectOnNoResponsePeriod = 10;
|
|
_mqttClient!.autoReconnect = true;
|
|
_mqttClient!.onDisconnected = _clientDisconnected;
|
|
_mqttClient!.onAutoReconnect = _clientDisconnected;
|
|
_mqttClient!.onAutoReconnected = _clientConnected;
|
|
|
|
var willProperties = MqttWillProperties();
|
|
willProperties.messageExpiryInterval = 3600;
|
|
willProperties.willDelayInterval = 0;
|
|
|
|
var connectionLostMessage = ConnectionLostMessage();
|
|
connectionLostMessage.connectionLost = true;
|
|
var connectionLostEncapsulatedMessage = _encapsulateMessage(const Uuid().v4(), false, connectionLostMessage.toJson());
|
|
|
|
_mqttClient!.connectionMessage = MqttConnectMessage()
|
|
.withClientIdentifier(Util.getAppUuid())
|
|
.authenticateAs("app", "apppwd")
|
|
.withSessionExpiryInterval(2700)
|
|
.withWillProperties(willProperties)
|
|
.withWillTopic("$_toServerTopic$_clientId/connection_lost")
|
|
.withWillQos(MqttQos.atMostOnce)
|
|
.keepAliveFor(60)
|
|
.withWillPayload(MqttUtf8Encoding().toUtf8(connectionLostEncapsulatedMessage))
|
|
.will();
|
|
}
|
|
|
|
await _mqttClient!.connect();
|
|
|
|
if (_mqttClient!.connectionStatus?.state == MqttConnectionState.connected) {
|
|
Util.logInfo('Verbunden mit dem MQTT-Broker!');
|
|
|
|
_reconnectDelay = 0;
|
|
|
|
await _setupMqtt();
|
|
|
|
await _initSession();
|
|
} else {
|
|
Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!');
|
|
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicConnectionStatusChanged,
|
|
payload: ConnectionStatus(status: ConnectionStatusEnum.notConnected),
|
|
);
|
|
}
|
|
} catch (ex) {
|
|
Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!: $ex');
|
|
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicConnectionStatusChanged,
|
|
payload: ConnectionStatus(status: ConnectionStatusEnum.notConnected),
|
|
);
|
|
}
|
|
}
|
|
|
|
bool _publishMessage(String topic, Map<String, dynamic> message, bool acknowledgeRequired, {bool enqueueOnError = true}) {
|
|
try {
|
|
String messageId = const Uuid().v4();
|
|
String messageText = jsonEncode(message);
|
|
|
|
if (acknowledgeRequired) {
|
|
_enqueueMessage(messageId, topic, messageText);
|
|
}
|
|
|
|
var encapsulatedMessageText = _encapsulateMessage(messageId, acknowledgeRequired, message);
|
|
|
|
var buffer = Uint8Buffer();
|
|
buffer.addAll(GZipEncoder().encode(const Utf8Encoder().convert(encapsulatedMessageText)) as Iterable<int>);
|
|
|
|
var messageTopic = "$_toServerTopic$_clientId/$topic";
|
|
|
|
_mqttClient!.publishMessage(messageTopic, MqttQos.atMostOnce, buffer);
|
|
|
|
Util.logInfo("message\n$encapsulatedMessageText\nsend to $messageTopic");
|
|
Util.logInfo("size in bytes: ${buffer.length}");
|
|
|
|
return true;
|
|
} catch (ex) {
|
|
Util.logInfo("send failed!");
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
Future<void> _setupMqtt() async {
|
|
Util.logInfo('_setupMqtt called');
|
|
|
|
_mqttClient!.updates.listen((List<MqttReceivedMessage<MqttMessage>> messages) {
|
|
for (var mqttMessage in messages) {
|
|
try {
|
|
final MqttPublishMessage message = mqttMessage.payload as MqttPublishMessage;
|
|
|
|
var topicArray = message.variableHeader?.topicName;
|
|
var topicArraySplited = topicArray!.split("/");
|
|
var topic = topicArraySplited[7];
|
|
|
|
Util.logInfo('message received\ntopic: $topic\n');
|
|
Util.logInfo("size in bytes: ${message.payload.message!.length.toString()}");
|
|
|
|
var decompressedPayload = utf8.decode(GZipDecoder().decodeBytes(message.payload.message!));
|
|
|
|
var headerMessage = MessageHeader.fromJson(jsonDecode(decompressedPayload));
|
|
Util.logInfo('$topic\nmessageId: ${headerMessage.messageId}');
|
|
|
|
var payloadString = jsonEncode(headerMessage.payload);
|
|
Util.logInfo('$topic\nplayloadString: $payloadString');
|
|
|
|
switch (topic) {
|
|
case 'session':
|
|
_sessionMessageReceived(jsonDecode(payloadString));
|
|
break;
|
|
|
|
case 'status':
|
|
_statusMessageReceived(jsonDecode(payloadString));
|
|
break;
|
|
|
|
case 'ranking':
|
|
_rankingMessageReceived(jsonDecode(payloadString));
|
|
break;
|
|
|
|
case 'job':
|
|
_jobMessageReceived(jsonDecode(payloadString));
|
|
break;
|
|
|
|
case 'job_del':
|
|
_jobDeleteReceived(jsonDecode(payloadString));
|
|
break;
|
|
|
|
case 'sync_request':
|
|
_syncRequestReceived(jsonDecode(payloadString));
|
|
break;
|
|
|
|
case 'location_status':
|
|
_locationStatusMessageReceived(jsonDecode(payloadString));
|
|
break;
|
|
|
|
case "mssg":
|
|
_messageMessageReceived(jsonDecode(payloadString));
|
|
break;
|
|
|
|
case "upload_device_log":
|
|
_messageUploadDeviceLogReceived(jsonDecode(payloadString));
|
|
break;
|
|
|
|
case "acknowledge":
|
|
_messageAcknowledgeReceived(jsonDecode(payloadString));
|
|
break;
|
|
|
|
default:
|
|
Util.logInfo('Unknown topic: $topic');
|
|
break;
|
|
}
|
|
} catch (ex) {
|
|
Util.logException(ex.toString());
|
|
}
|
|
}
|
|
});
|
|
|
|
var fromServerTopic = "$_fromServerTopic$_clientId/#";
|
|
|
|
if (Util.preferencesContainsKey(Constants.prefsPreviousMqttSubscription)) {
|
|
String previousMqttSubscription = Util.preferencesGetString(Constants.prefsPreviousMqttSubscription, "") as String;
|
|
|
|
if (previousMqttSubscription != fromServerTopic) {
|
|
Util.logInfo('topic changed -> unsubscribe from $previousMqttSubscription');
|
|
_mqttClient!.unsubscribeStringTopic(previousMqttSubscription);
|
|
}
|
|
|
|
_mqttClient!.subscribe(_fromServerTopic, MqttQos.exactlyOnce);
|
|
}
|
|
|
|
_mqttClient!.subscribe(fromServerTopic, MqttQos.exactlyOnce);
|
|
Util.preferencesSaveString(Constants.prefsPreviousMqttSubscription, fromServerTopic);
|
|
|
|
Util.logInfo('subscribed to topic: $fromServerTopic');
|
|
|
|
// TODO MESSAGING
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicConnectionStatusChanged,
|
|
payload: ConnectionStatus(status: ConnectionStatusEnum.connected),
|
|
);
|
|
|
|
//await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
|
|
}
|
|
|
|
void _sessionMessageReceived(payloadData) async {
|
|
connectionState = ConnectionStateEnum.connected;
|
|
|
|
await _sendSyncMessage();
|
|
|
|
await _sendEnqueuedMessages();
|
|
|
|
var sessionStatusMessage = SessionStatus.fromJson(payloadData);
|
|
|
|
await MyApp.isarInstance.writeTxn(() async {
|
|
await MyApp.isarInstance.sessionStatus.clear();
|
|
await MyApp.isarInstance.sessionStatus.put(sessionStatusMessage);
|
|
});
|
|
|
|
// TODO MESSAGING
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicSessionStatusInfoUpdate,
|
|
payload: SessionStatusInfoUpdate(status: sessionStatusMessage),
|
|
);
|
|
|
|
//await MessageService().publish<SessionStatusInfoUpdate>(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);
|
|
});
|
|
|
|
// TODO MESSAGING
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicStatusInfoUpdated,
|
|
payload: StatusInfoUpdate(status: statusMessage),
|
|
);
|
|
|
|
//await MessageService().publish<StatusInfoUpdate>(Constants.messageTopicStatusInfoUpdated, arguments: StatusInfoUpdate(status: statusMessage));
|
|
}
|
|
|
|
void _locationStatusMessageReceived(payloadData) async {
|
|
var locationStatus = LocationStatus.fromJson(payloadData);
|
|
|
|
await MyApp.isarInstance.writeTxn(() async {
|
|
await MyApp.isarInstance.locationStatus.clear();
|
|
await MyApp.isarInstance.locationStatus.put(locationStatus);
|
|
});
|
|
|
|
// TODO MESSAGING
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicLocationStatusUpdated,
|
|
payload: LocationStatusUpdate(status: locationStatus),
|
|
);
|
|
|
|
// await MessageService().publish<LocationStatusUpdate>(Constants.messageTopicLocationStatusUpdated, arguments: LocationStatusUpdate(status: locationStatus));
|
|
}
|
|
|
|
void _rankingMessageReceived(payloadData) async {
|
|
var ranking = Ranking.fromJson(payloadData);
|
|
|
|
await MyApp.isarInstance.writeTxn(() async {
|
|
await MyApp.isarInstance.rankings.clear();
|
|
await MyApp.isarInstance.rankings.put(ranking);
|
|
});
|
|
|
|
// TODO MESSAGING
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicRankingUpdated,
|
|
payload: RankingUpdated(),
|
|
);
|
|
|
|
//await MessageService().publish<RankingUpdated>(Constants.messageTopicRankingUpdated, arguments: RankingUpdated());
|
|
}
|
|
|
|
void _jobMessageReceived(payloadData) async {
|
|
Job newJob = Job.fromJson(payloadData);
|
|
|
|
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');
|
|
}
|
|
}
|
|
|
|
_acknowlegdeJob(newJob);
|
|
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicJobReceived,
|
|
payload: JobReceivedMessage(job: newJob),
|
|
);
|
|
}
|
|
|
|
void _jobDeleteReceived(payloadData) async {
|
|
var jobDelMessage = JobDelMessage.fromJson(payloadData);
|
|
|
|
var job = await MyApp.isarInstance.jobs.where().filter().idEqualTo(jobDelMessage.id).findFirst();
|
|
|
|
await MyApp.isarInstance.writeTxn(() async {
|
|
await MyApp.isarInstance.jobs.where().filter().idEqualTo(jobDelMessage.id).deleteAll();
|
|
});
|
|
|
|
if (job != null) {
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicJobDeleted,
|
|
payload: JobDeletedMessage(job: job),
|
|
);
|
|
}
|
|
}
|
|
|
|
void _syncRequestReceived(payloadData) async {
|
|
Util.logInfo('_syncRequestReceived');
|
|
|
|
await _sendEnqueuedMessages();
|
|
|
|
await _sendSyncMessage();
|
|
}
|
|
|
|
void _messageMessageReceived(Map<String, dynamic> payloadData) async {
|
|
var newMessage = await ChatMessage.mergeFromJson(payloadData);
|
|
|
|
await MyApp.isarInstance.writeTxn(() async {
|
|
await MyApp.isarInstance.chatMessages.put(newMessage);
|
|
});
|
|
|
|
await Util.playMp3('asbeep.mp3');
|
|
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicMessageReceived,
|
|
payload: MessageReceivedMessage(message: newMessage),
|
|
);
|
|
}
|
|
|
|
void _messageUploadDeviceLogReceived(payloadData) async {
|
|
var deviceLogRequest = DeviceLogRequest.fromJson(payloadData);
|
|
|
|
var logEintries = await MyApp.isarInstance.logEntrys
|
|
.filter()
|
|
.timeGreaterThan(deviceLogRequest.startDateTime)
|
|
.timeLessThan(deviceLogRequest.endDateTime)
|
|
.findAll();
|
|
|
|
var deviceLog = DeviceLogResponse();
|
|
deviceLog.entries = logEintries;
|
|
|
|
_publishMessage("device_log", deviceLog.toJson(), false);
|
|
}
|
|
|
|
void _messageAcknowledgeReceived(payloadData) async {
|
|
var acknowledgeMessage = AcknowledgeMessage.fromJson(payloadData);
|
|
|
|
await MyApp.isarInstance.writeTxn(() async {
|
|
await MyApp.isarInstance.enqueuedMessages
|
|
.filter()
|
|
.messageIdEqualTo(acknowledgeMessage.messageId)
|
|
.deleteAll();
|
|
});
|
|
}
|
|
|
|
|
|
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 (!_disconnectInProgress) {
|
|
if (_reconnectTimer != null && _reconnectTimer!.isActive) {
|
|
_reconnectTimer!.cancel();
|
|
}
|
|
}
|
|
}
|
|
|
|
Future<void> _clientDisconnected() async {
|
|
Util.logInfo("_clientDisconnected called");
|
|
|
|
connectionState = ConnectionStateEnum.notConnected;
|
|
|
|
_cancelTrackTimer();
|
|
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicConnectionStatusChanged,
|
|
payload: ConnectionStatus(status: ConnectionStatusEnum.notConnected),
|
|
);
|
|
}
|
|
|
|
Future<void> _clientConnected() async {
|
|
Util.logInfo("_clientConnected called");
|
|
|
|
connectionState = ConnectionStateEnum.connected;
|
|
|
|
_initSession();
|
|
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicConnectionStatusChanged,
|
|
payload: ConnectionStatus(status: ConnectionStatusEnum.connected),
|
|
);
|
|
}
|
|
|
|
void _processBootstrap() async {
|
|
try {
|
|
var deviceInfoPlugin = DeviceInfoPlugin();
|
|
|
|
AndroidDeviceInfo androidInfo = await deviceInfoPlugin.androidInfo;
|
|
BaseDeviceInfo deviceInfo = await deviceInfoPlugin.deviceInfo;
|
|
|
|
var deviceObject = DeviceObject();
|
|
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 = Util.getAppUuid();
|
|
handshakeMessage.deviceObject = deviceObject;
|
|
handshakeMessage.ver = await Util.getAppVersion();
|
|
|
|
var jsonMap = handshakeMessage.toJson();
|
|
var json = Uri.encodeFull(jsonEncode(jsonMap));
|
|
var uri = "https://bootstrap.assecutor.de/bootstrap?json=$json";
|
|
|
|
Util.logInfo("Bootstrap request: $uri");
|
|
|
|
var response = await http.get(Uri.parse(uri));
|
|
var responseText = response.body;
|
|
Util.logInfo("Bootstrap response: $responseText");
|
|
|
|
var bootstrapResponsePayload = BootstrapResponse.fromJson(jsonDecode(responseText));
|
|
|
|
var status = bootstrapResponsePayload.status;
|
|
var message = bootstrapResponsePayload.message;
|
|
|
|
if (status > 0) {
|
|
showAlertDialog(status == 1 ? "Information" : "Fehler", message);
|
|
}
|
|
|
|
if (status < 2) {
|
|
var brokerUriAndPortArray = bootstrapResponsePayload.broker.replaceFirst("mqtt://", "").split(":");
|
|
_brokerUri = brokerUriAndPortArray[0];
|
|
_brokerPort = int.parse(brokerUriAndPortArray[1]);
|
|
_clientId = bootstrapResponsePayload.id;
|
|
_fromServerTopic = bootstrapResponsePayload.fromServer;
|
|
_toServerTopic = bootstrapResponsePayload.toServer;
|
|
|
|
_connectToBroker();
|
|
}
|
|
} catch (ex) {
|
|
Util.logException("bootstrap failed: {$ex}");
|
|
|
|
_startReconnectTimer();
|
|
}
|
|
}
|
|
|
|
_initSession() async {
|
|
var startSessionMessage = StartSessionMessage();
|
|
startSessionMessage.desiredZip = (Util.preferencesGetString(Constants.prefsDesiredZip, ""))!;
|
|
|
|
var jsonMap = startSessionMessage.toJson();
|
|
|
|
_publishMessage("session", jsonMap, false);
|
|
}
|
|
|
|
_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);
|
|
}
|
|
|
|
_publishMessage("sync", answer.toJson(), false);
|
|
}
|
|
|
|
void _startTrackTimer() {
|
|
_trackTimer ??= Timer.periodic(const Duration(seconds: 120), (Timer t) {
|
|
try {
|
|
var position = MyApp.currentPosition;
|
|
if (position != null) {
|
|
var trackMessage = TrackPosition();
|
|
trackMessage.time = position.timestamp.toIso8601String();
|
|
trackMessage.lat = position.latitude;
|
|
trackMessage.long = position.longitude;
|
|
trackMessage.acc = position.accuracy.toInt();
|
|
|
|
_publishMessage("track", trackMessage.toJson(), false);
|
|
}
|
|
} catch (ex) {
|
|
Util.logException("Error in periodic track timer: {$ex}");
|
|
}
|
|
});
|
|
}
|
|
|
|
void _cancelTrackTimer() {
|
|
if (_trackTimer != null && _trackTimer!.isActive) {
|
|
_trackTimer!.cancel();
|
|
_trackTimer = null;
|
|
}
|
|
}
|
|
|
|
void _enqueueMessage(String uuid, String receiver, String payload) async {
|
|
var enqueuedMessage = EnqueuedMessage(messageId: uuid);
|
|
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 {
|
|
var json = jsonDecode(enqueuedMessage.payload);
|
|
|
|
if (_publishMessage(enqueuedMessage.topic, json, false, enqueueOnError: false)) {
|
|
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 = AcknowledgeJobReceived();
|
|
message.id = job.id;
|
|
message.state = job.state;
|
|
message.ver = job.ver;
|
|
message.utc = DateTime.now().millisecondsSinceEpoch;
|
|
|
|
_publishMessage('job_received', message.toJson(), false);
|
|
}
|
|
|
|
void initMessages() {
|
|
subscribe(
|
|
queueId: Constants.messageTopicPublishPayload,
|
|
callback: (message) {
|
|
var publishPayloadMessage = message.payload as PublishPayloadMessage;
|
|
|
|
_publishMessage(publishPayloadMessage.topic, publishPayloadMessage.message, publishPayloadMessage.acknowledgeRequired);
|
|
},
|
|
);
|
|
|
|
subscribe(
|
|
queueId: Constants.messageTopicDisconnectClient,
|
|
callback: (message) {
|
|
_disconnectMqttClient();
|
|
},
|
|
);
|
|
|
|
subscribe(
|
|
queueId: Constants.messageTopicConnectClient,
|
|
callback: (message) {
|
|
_connectToBroker();
|
|
},
|
|
);
|
|
|
|
subscribe(
|
|
queueId: Constants.messageTopicRequestConnectionStatus,
|
|
callback: (message) {
|
|
bool connected = false;
|
|
if (_mqttClient != null && _mqttClient!.connectionStatus != null) {
|
|
if (_mqttClient!.connectionStatus!.state == MqttConnectionState.connected) {
|
|
connected = true;
|
|
}
|
|
}
|
|
|
|
sendMessage(
|
|
routingKey: Constants.messageTopicCurrentConnectionStatus,
|
|
payload: CurrentConnectionStatusMessage(state: connected),
|
|
);
|
|
},
|
|
);
|
|
}
|
|
|
|
String _encapsulateMessage(String messageId, bool requiresAck, Map<String, dynamic> message) {
|
|
var messageHeader = MessageHeader();
|
|
messageHeader.messageId = messageId;
|
|
messageHeader.requiresAck = requiresAck;
|
|
messageHeader.payload = message;
|
|
|
|
var encapsulatedMessageText = jsonEncode(messageHeader.toJson());
|
|
|
|
return encapsulatedMessageText;
|
|
}
|
|
}
|