UniRxを用いたSerialHandler

Unity + Arduinoな構成のインスタレーションを作ることになり、その際UniRxを採用して詰まったことのメモです。

TL;DR

• Thread 内では ReactiveProperty のValueを更新できない(イベントの購読が出来ない)
• その対策として、一度メンバ変数に値を渡し、ObserveEveryValueChangedと ToReadOnlyReactiveProperty を用いることで、間接的に ReactiveProperty にすることにした。

Thread内ではReactivePropertyのValueは更新できない

通常Arduinoの値をSerialから読み取る際、メインスレッドの速度に影響が出ないよう、 Thread を用いてそのThread内でSerialの読み取りを行います。
その読み取った値を ReactiveProperty として扱いたくこのような処理を書きました。

SerialHandler.cs

// 説明用のため、細かいところは省略してあります。

private ReactiveProperty<string> _sensorMsg = new ReactiveProperty<string>();
public IReadOnlyReactiveProperty<string> SensorMsg { get { return _sensorMsg; } }

void Awake () {
  _serialPort = new SerialPort (_portName, _baudRate, Parity.None, 8, StopBits.One);
  _serialPort.Open();

  _thread = new Thread (Read);
  _thread.Start();
}

private void Read () {
  while (_isRunning && _serialPort != null && _serialPort.IsOpen) {
    try {
      _sensorMsg.Value = _serialPort.ReadLine();
    }
    catch (System.Exception e) {
      Debug.LogWarning (e.Message);
    }
  }
}

しかし、このまま SensorMsg を購読しようとすると、実行時にエラーとなり、値を受け取ることは出来ません。
これはThread内では、UnityのAPIを実行できないという性質によるものかと思われます。

ObserveEveryValueChanged を使用する

Thread内で更新できないものはどうしようもないので、一度メンバ変数に値を渡すことにしました。
その上で、 ObserveEveryValueChanged というUniRxの関数を使うことにしました。
この ObserveEveryValueChanged は任意のクラスオブジェクトの変更をストリームに変換できる事のできる関数です。
こちらの関数と、 そのストリームをreadonlyな ReactiveProperty に変換する、 ToReadOnlyReactiveProperty を使って処理を書きました。

完成したコード

SerialHandler.cs

using System.Collections.Generic;
using System.IO.Ports;
using System.Threading;
using SimpleJSON;
using UniRx;
using UnityEngine;


public class SerialHandler : MonoBehaviour {
  [SerializeField] private string _portName;

  private SerialPort _serialPort;
  private Thread _thread;
  private bool _isRunning;

  private int _baudRate = 115200;

  private string _sensorMsg;

  public IReadOnlyReactiveProperty<string> SensorMsg { get; private set; }

  void Awake () {
    _serialPort = new SerialPort (_portName, _baudRate, Parity.None, 8, StopBits.One);
    _serialPort.Open();

    _serialPort.DtrEnable = true;
    _serialPort.RtsEnable = true;
    _serialPort.DiscardInBuffer();

    _isRunning = true;

    _thread = new Thread (Read);
    _thread.Start();

    SensorMsg = this.ObserveEveryValueChanged (x => x._sensorMsg)
      .ToReadOnlyReactiveProperty();
  }

  private void Read () {
    while (_isRunning && _serialPort != null && _serialPort.IsOpen) {
      try {
        _sensorMsg = _serialPort.ReadLine();
      }
      catch (System.Exception e) {
        Debug.LogWarning (e.Message);
      }
    }
  }

  public void Close () {
    _isRunning = false;

    if (_thread != null && _thread.IsAlive) {
      _thread.Join();
    }

    if (_serialPort != null && _serialPort.IsOpen) {
      _serialPort.Close();
      _serialPort.Dispose();
    }
  }
}