Worker Decode

worker_decode.dart

import 'dart:async';
import 'dart:convert';
import 'dart:isolate';

import 'package:flutter/foundation.dart';
import 'package:rxdart/rxdart.dart';

class WorkerServiceTask {
  final ReceivePort _receivePort;
  final SendPort _sendPort;

  WorkerServiceTask(this._receivePort, this._sendPort);

  static Future<WorkerServiceTask> spawn() async {
    final initPort = RawReceivePort();
    final receivedPortConnection = Completer<ReceivePort>();
    final sendPortConnection = Completer<SendPort>();
    initPort.handler = (initMessage) {
      final commandPort = initMessage as SendPort;
      receivedPortConnection.complete(ReceivePort.fromRawReceivePort(initPort));
      sendPortConnection.complete(commandPort);
    };
    try {
      await Isolate.spawn(_remoteWorker, initPort.sendPort);
    } on Object {
      initPort.close();
    }
    final receivePortFuture = receivedPortConnection.future;
    final sendPortFuture = sendPortConnection.future;

    return WorkerServiceTask(await receivePortFuture, await sendPortFuture);
  }

  static void _remoteWorker(SendPort sendPort) {
    final receivePort = ReceivePort();
    sendPort.send(receivePort.sendPort);
    _listenWorker(receivePort, receivePort.sendPort);
  }

  static void _listenWorker(ReceivePort receivePort, SendPort sendPort) {
    receivePort.listen(
      (message) {
        if (message is Map<WorkerIsolateTypes, dynamic>) {
          if (message.containsKey(WorkerIsolateTypes.decode)) {
            final String? jsonStringItem = message[WorkerIsolateTypes.decode];
            if (jsonStringItem != null) _decodeJson(jsonStringItem, sendPort);
          }
        } else if (message is WorkerIsolateTypes &&
            message == WorkerIsolateTypes.closed) {
          receivePort.close();
        } else {
          sendPort.send(ArgumentError());
        }
      },
    );
  }

  static void _decodeJson(String value, SendPort sendPort) {
    try {
      sendPort.send(_decode(value));
    } catch (error, stackTrace) {
      sendPort.send(RemoteError(error.toString(), stackTrace.toString()));
    }
  }

  static Map<String, dynamic> _decode(String value) {
    final Map<String, dynamic> decodeJsonResult = jsonDecode(value);
    return decodeJsonResult;
  }

  Future<Map<String, dynamic>> decodeResultItem() {
    return _receivePort
        .map((event) {
          if (event is Map<String, dynamic>) {
            return event;
          } else if (event is RemoteError) {
            throw Exception(event.stackTrace);
          } else {
            return null;
          }
        })
        .whereNotNull()
        .first;
  }

  Stream<Map<String, dynamic>> watchDecodeListener() => _receivePort
      .map(
        (event) {
          if (event is Map<String, dynamic>) {
            return event;
          } else if (event is RemoteError) {
            throw Exception(event.stackTrace);
          } else {
            return null;
          }
        },
      )
      .whereNotNull()
      .asBroadcastStream();

  void onSendDataJsonToWorker(WorkerIsolateTypes type, String value) {
    _sendPort.send(<WorkerIsolateTypes, String>{type: value});
  }

  void onClose() => _sendPort.send(WorkerIsolateTypes.closed);
}

enum WorkerIsolateTypes {
  decode,
  closed,
}