378 lines
12 KiB
Dart
378 lines
12 KiB
Dart
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';
|
|
import 'package:stadtbote/protocoll_messages/bootstrap_request.dart';
|
|
import 'package:stadtbote/protocoll_messages/bootstrap_response_payload.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/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/tools/connection_status.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/session_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/status.dart';
|
|
import 'constants.dart';
|
|
import 'util.dart';
|
|
|
|
enum ConnectionStateEnum {
|
|
not_connected,
|
|
connected
|
|
}
|
|
|
|
class MqttCore {
|
|
static late MqttCore instance;
|
|
|
|
late ConnectionStateEnum connectionState = ConnectionStateEnum.not_connected;
|
|
|
|
late String _brokerUri;
|
|
late int _brokerPort;
|
|
late String _fromServerTopic;
|
|
late String _toServerTopic;
|
|
|
|
late MqttServerClient _mqttClient;
|
|
|
|
late int _reconnectDelay = 0;
|
|
|
|
MqttCore() {
|
|
_processBootstrap();
|
|
}
|
|
|
|
static setup() {
|
|
instance = MqttCore();
|
|
}
|
|
|
|
_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!');
|
|
|
|
BroadcastReceiver().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
|
|
|
|
_startReconnectTimer();
|
|
}
|
|
|
|
} catch (ex) {
|
|
_startReconnectTimer();
|
|
|
|
Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!: $ex');
|
|
|
|
BroadcastReceiver().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
|
|
}
|
|
}
|
|
|
|
publishText(String topic, String text) {
|
|
try {
|
|
var buffer = Uint8Buffer();
|
|
buffer.addAll(text.codeUnits);
|
|
|
|
_mqttClient.publishMessage(topic, MqttQos.exactlyOnce, buffer);
|
|
|
|
Util.logInfo("message $text send to $topic");
|
|
} catch (e) {
|
|
Util.logException(e.toString());
|
|
}
|
|
}
|
|
|
|
publishPayloadMessage(String receiver, Map<String, dynamic> payload) {
|
|
try {
|
|
var payloadContainer = PayloadContainer();
|
|
payloadContainer.id = const Uuid().v4().toString();
|
|
payloadContainer.recv = receiver;
|
|
payloadContainer.store = true;
|
|
payloadContainer.payload = payload;
|
|
|
|
var json = jsonEncode(payloadContainer.toJson());
|
|
|
|
publishText(_toServerTopic, json);
|
|
} catch (e) {
|
|
Util.logException(e.toString());
|
|
}
|
|
}
|
|
|
|
void _setupSubscriptions() async {
|
|
Util.logInfo('_setupSubscriptions called');
|
|
|
|
_mqttClient.updates?.listen((List<MqttReceivedMessage<MqttMessage>> c) {
|
|
try {
|
|
final MqttPublishMessage message = c[0].payload as MqttPublishMessage;
|
|
final payload = MqttPublishPayload.bytesToStringAsString(message.payload.message);
|
|
var topic = message.variableHeader?.topicName;
|
|
|
|
Util.logInfo('message received');
|
|
Util.logInfo('topic: $topic');
|
|
Util.logInfo('payload: $payload');
|
|
|
|
Map<String, dynamic> messageData = jsonDecode(payload);
|
|
|
|
for (var entry in messageData.keys) {
|
|
if (entry == 'payload') {
|
|
var payload = messageData[entry];
|
|
for (var payloadEntry in payload.keys) {
|
|
var payloadData = payload[payloadEntry];
|
|
|
|
switch(payloadEntry) {
|
|
case 'session':
|
|
_sessionMessageReceived(payloadData);
|
|
break;
|
|
|
|
case 'ranks':
|
|
_rankingMessageReceived(payloadData);
|
|
break;
|
|
|
|
case 'jobs':
|
|
_jobMessageReceived(payloadData);
|
|
break;
|
|
|
|
case 'job_del':
|
|
//_jobDeleteReceived(payloadData);
|
|
break;
|
|
|
|
case 'sync_request':
|
|
_syncRequestReceived(payloadData);
|
|
break;
|
|
|
|
case "mssg":
|
|
_messageMessageReceived(payloadData);
|
|
break;
|
|
|
|
default:
|
|
Util.logInfo('Unknown topic: $payloadEntry');
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
} catch(ex) {
|
|
Util.logException(ex.toString());
|
|
}
|
|
});
|
|
|
|
_mqttClient.subscribe(_fromServerTopic, MqttQos.exactlyOnce);
|
|
Util.logInfo('subscribed to topic: $_fromServerTopic');
|
|
|
|
BroadcastReceiver().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
|
|
}
|
|
|
|
void _sessionMessageReceived(payloadData) async {
|
|
connectionState = ConnectionStateEnum.connected;
|
|
|
|
await _sendSyncMessage();
|
|
|
|
var statusMessage = Status.fromJson(payloadData['status']);
|
|
|
|
await MyApp.isarInstance.writeTxn(() async {
|
|
await MyApp.isarInstance.status.put(statusMessage);
|
|
});
|
|
|
|
BroadcastReceiver().publish<StatusInfoUpdate>(Constants.messageTopicStatusInfoUpdate, arguments: StatusInfoUpdate(status: statusMessage));
|
|
}
|
|
|
|
void _rankingMessageReceived(payloadData) async {
|
|
Util.logInfo('_rankingMessageReceived');
|
|
|
|
/*var statusMessage = Status.fromJson(payloadData['status']);
|
|
|
|
await MyApp.isarInstance.writeTxn(() async {
|
|
await MyApp.isarInstance.status.put(statusMessage);
|
|
});
|
|
|
|
BroadcastReceiver().publish<StatusInfoUpdate>(Constants.messageTopicStatusInfoUpdate, arguments: StatusInfoUpdate(status: statusMessage));*/
|
|
}
|
|
|
|
void _jobMessageReceived(payloadData) async {
|
|
for (var foundJobData in payloadData) {
|
|
Job newJob = Job.fromJson(foundJobData);
|
|
|
|
await MyApp.isarInstance.writeTxn(() async {
|
|
await MyApp.isarInstance.jobs.put(newJob);
|
|
});
|
|
|
|
BroadcastReceiver().publish<JobReceivedMessage>(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob));
|
|
}
|
|
}
|
|
|
|
void _jobDeleteReceived(List<dynamic> payloadData) async {
|
|
Map<String, dynamic> map = {};
|
|
map["jobDel"] = payloadData;
|
|
var jobDelMessage = JobDelMessage.fromJson(map);
|
|
|
|
for(var jobId in jobDelMessage.jobDel) {
|
|
await MyApp.isarInstance.writeTxn(() async {
|
|
await MyApp.isarInstance.jobs.delete(jobId);
|
|
});
|
|
|
|
BroadcastReceiver().publish<JobDeletedMessage>(Constants.messageTopicJobDeleted, arguments: JobDeletedMessage(jobId: jobId));
|
|
}
|
|
}
|
|
|
|
void _syncRequestReceived(payloadData) async {
|
|
Util.logInfo('_syncRequestReceived');
|
|
|
|
/*for (var foundJobData in payloadData) {
|
|
Job newJob = Job.fromJson(foundJobData);
|
|
|
|
await MyApp.isarInstance.writeTxn(() async {
|
|
await MyApp.isarInstance.jobs.put(newJob);
|
|
});
|
|
|
|
BroadcastReceiver().publish<JobReceivedMessage>(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob));
|
|
}*/
|
|
}
|
|
|
|
void _messageMessageReceived(Map<String, dynamic> payloadData) async {
|
|
var newMessage = Message.fromJson(payloadData);
|
|
|
|
await MyApp.isarInstance.writeTxn(() async {
|
|
await MyApp.isarInstance.messages.put(newMessage);
|
|
});
|
|
|
|
BroadcastReceiver().publish<MessageReceivedMessage>(Constants.messageTopicMessageReceived, arguments: MessageReceivedMessage(message: newMessage));
|
|
}
|
|
|
|
void _startReconnectTimer() {
|
|
connectionState = ConnectionStateEnum.not_connected;
|
|
|
|
Util.logInfo("_startReconnectTimer called");
|
|
|
|
_reconnectDelay = min(_reconnectDelay + 1, 30);
|
|
|
|
Timer(Duration(seconds: _reconnectDelay), () {
|
|
_processBootstrap();
|
|
});
|
|
}
|
|
|
|
void _clientDisconnected() {
|
|
connectionState = ConnectionStateEnum.not_connected;
|
|
|
|
Util.logInfo("_clientDisconnected called");
|
|
|
|
BroadcastReceiver().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
|
|
|
|
_startReconnectTimer();
|
|
}
|
|
|
|
void _processBootstrap() async {
|
|
try {
|
|
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 = "https://init-x.assecutor.de/bootstrap?get_mqtt_broker=$json";
|
|
Util.logInfo("Bootstrap request: $uri");
|
|
|
|
var response = await http.get(Uri.parse(uri));
|
|
var responseText = response.body;
|
|
|
|
var bootstrapResponsePayload = BootstrapResponsePayload.fromJson(jsonDecode(responseText)["payload"]);
|
|
|
|
var brokerUriAndPortArray = bootstrapResponsePayload.config.broker.replaceFirst("mqtt://", "").split(":");
|
|
_brokerUri = brokerUriAndPortArray[0];
|
|
_brokerPort = int.parse(brokerUriAndPortArray[1]);
|
|
_fromServerTopic = bootstrapResponsePayload.config.topics.fromServer;
|
|
_toServerTopic = bootstrapResponsePayload.config.topics.toServer;
|
|
|
|
_connectToBroker();
|
|
} catch (ex) {
|
|
_startReconnectTimer();
|
|
}
|
|
}
|
|
|
|
_initSession() {
|
|
var playload = Payload();
|
|
playload.startSession = "true";
|
|
|
|
var startSessionMessage = StartSessionMessage();
|
|
startSessionMessage.recv = "server";
|
|
startSessionMessage.store = true;
|
|
startSessionMessage.messageId = const Uuid().v4();
|
|
startSessionMessage.payload = playload;
|
|
|
|
var jsonMap = startSessionMessage.toJson();
|
|
var json = jsonEncode(jsonMap);
|
|
Util.logInfo("send start session json: $json");
|
|
|
|
final builder = MqttClientPayloadBuilder();
|
|
builder.addString(json);
|
|
var payload = builder.payload;
|
|
|
|
var topic = "$_toServerTopic/init";
|
|
Util.logInfo("send start session json to topic: $topic");
|
|
|
|
var messageId = _mqttClient.publishMessage(topic, MqttQos.exactlyOnce, payload!);
|
|
Util.logInfo("send start session json with messageId: $messageId");
|
|
}
|
|
|
|
_sendSyncMessage() async
|
|
{
|
|
var jobs = await MyApp.isarInstance.jobs.filter().stateGreaterThan(-1).findAll();
|
|
|
|
var answer = SyncJobsMessage();
|
|
answer.sync = SyncJobMessageSync();
|
|
answer.sync.jobs = [];
|
|
answer.sync.utcUnixTime = DateTime.now().toUtc().millisecondsSinceEpoch ~/ 1000;
|
|
for (var job in jobs) {
|
|
SyncJobMessageJob syncJob = SyncJobMessageJob();
|
|
syncJob.state = job.state;
|
|
syncJob.id = job.id;
|
|
syncJob.ver = job.ver;
|
|
answer.sync.jobs.add(syncJob);
|
|
}
|
|
|
|
MqttCore.instance.publishPayloadMessage("server", answer.toJson());
|
|
}
|
|
}
|