reactive4java で端末の方位を取得しつづける #Java

Reactive Extensions いいよ Reactive Extensions。
Reactive Extensions を Java に移植したのが reactive4java。
.NET じゃないので Linq やラムダ式は使えない(いやスクリプトとしてなら使える)けど、並列処理を直列に書けるヨロコビは味わえます。

Android でも使えます。
Android の端末の方位を通知する処理を書いてみました(仕事で必要だったので)。

SensorFunctions.java

import hu.akarnokd.reactive4java.base.Func1;
import hu.akarnokd.reactive4java.base.Functions;
import hu.akarnokd.reactive4java.reactive.Observable;
import hu.akarnokd.reactive4java.reactive.Observer;
import hu.akarnokd.reactive4java.reactive.Reactive;

public final class SensorFunctions {

    /**
     * 端末の確度を Observable として取得する
     * 
     * ※端末は縦向き前提です。横向きには対応していません。
     * port from sato-c / sensorcheck https://github.com/sato-c/sensorcheck
     */
    static public Observable<Float> getRotationAsObservable(final Context context, final int rate) {
        return Reactive.createWithCloseable(new Func1<Observer<? super Float>, Closeable>() {

            @Override
            public Closeable invoke(final Observer<? super Float> observer) {
                final AtomicBoolean _stop = new AtomicBoolean(false); // 停止フラグ

                final SensorManager accelMan = (SensorManager)context.getSystemService(Context.SENSOR_SERVICE);
                final SensorManager magMan   = (SensorManager)context.getSystemService(Context.SENSOR_SERVICE);

                final SensorEventListener listener = new SensorEventListener() {
                    final static private int _MATRIX_SIZE = 16;
                    final static private int _XYZ_AXIS = 3;
                    private boolean _sensorReady = false;
                    private float[] _R = new float[_MATRIX_SIZE];
                    private float[] _I = new float[_MATRIX_SIZE];
                    private float[] _magnetic = null;
                    private float[] _accel = null;
                    private float[] _orient = new float[_XYZ_AXIS];

                    /*
                     *  http://developer.android.com/reference/android/hardware/SensorEvent.html#values
                     */
                    @Override
                    public void onSensorChanged(SensorEvent event) {
                        // 停止されていたら読み飛ばす
                        if (_stop.get()) {
                            return;
                        }

                        switch ( event.sensor.getType() ) {
                        case    Sensor.TYPE_ACCELEROMETER:
                            _accel = event.values.clone();
                            break;
                        case    Sensor.TYPE_MAGNETIC_FIELD:
                            _sensorReady = true;
                            _magnetic = event.values.clone();
                            break;
                        }

                        if ( _sensorReady && _magnetic != null && _accel != null ) {
                            _sensorReady = false;

                            SensorManager.getRotationMatrix(_R, _I, _accel, _magnetic);
                            SensorManager.getOrientation(_R, _orient);

                            if ( _orient[0] < 0 ) {
                                _orient[0] = (float) (Math.toDegrees(_orient[0]) + 360.0f);
                            } else {
                                _orient[0] = (float) (Math.toDegrees(_orient[0]));
                            }

                            // 発火
                            observer.next(_orient[0]);
                        }
                    }

                    @Override
                    public void onAccuracyChanged(Sensor sensor, int accuracy) {
                    }
                };

                Sensor accelSensor = getSensorOnce(accelMan, Sensor.TYPE_ACCELEROMETER);
                if (accelSensor == null) {
                    observer.error(new InvalidParameterException("Sensor.TYPE_ACCELEROMETER not found."));
                    return Functions.EMPTY_CLOSEABLE;
                }

                Sensor magSensor = getSensorOnce(accelMan, Sensor.TYPE_MAGNETIC_FIELD);
                if (magSensor == null) {
                    observer.error(new InvalidParameterException("Sensor.TYPE_MAGNETIC_FIELD not found."));
                    return Functions.EMPTY_CLOSEABLE;
                }

                accelMan.registerListener(listener, accelSensor, rate);
                magMan.registerListener(listener, magSensor, rate);

                return new Closeable() {
                    @Override
                    public void close() throws IOException {
                        // 受信はすぐに止まらないのでフラグで読み飛ばす
                        if (_stop.get()) {
                            return;
                        }
                        _stop.set(true);

                        magMan.unregisterListener(listener);
                        accelMan.unregisterListener(listener);
                    }
                };
            }

            private Sensor getSensorOnce(SensorManager sensorMan, int sensorType) {
                List<Sensor> sensors = sensorMan.getSensorList(sensorType);
                if (sensors.size() == 0) {
                    return null;
                }

                return sensors.get(0);
            }
        });
    }
}

使うほうはこんな感じ。

Closeable closer = 
    SensorFunctions.getRotationAsObservable(
        context, SensorManager.SENSOR_DELAY_NORMAL))
.register(new Observer<Float>() { // Listen 開始
    @Override
    public void next(Float value) {
        // 角度が変更されると呼ばれる
        Log.d(TAG, "next() called. - " + value);
    }

    @Override
    public void finish() {
        // 終了(Closeable.Close) すると呼ばれる
        Log.d(TAG, "finish() called.");
    }

    @Override
    public void error(Throwable err) {
        // なんかエラー
        Log.e(TAG, "error() called.", err);
    }
});

Thread.sleep(3000); // 3秒待って

closer.Close(); // Listen 終了

.register() すると、角度が変わる度に、 next() が呼ばれ、角度が通知されます。
（処理自体は非同期で行われるので、上の例だと Sleep を挟まないとすぐに Listen 停止してしまいます。）

これ実際に実行すると、すごい勢いで next が呼ばれて大変なことになります。

これじゃデータ多すぎるよぉ・・・そんな時も Reactive Extensions は便利。

こうします。

Closeable closer = 
ObservableBuilder.from( // メソッドチェーンできるおまじない
    SensorFunctions.getRotationAsObservable(
        context, SensorManager.SENSOR_DELAY_NORMAL)))
.sample(1, TimeUnit.SECONDS) // 1秒置きの最新値にフィルタ
.register(new Observer<Float>() { // Listen 開始
    @Override
    public void next(Float value) {
        // 角度が変更されると呼ばれる
        Log.d(TAG, "next() called. - " + value);
    }

    @Override
    public void finish() {
        // 終了(Closeable.Close) すると呼ばれる
        Log.d(TAG, "finish() called.");
    }

    @Override
    public void error(Throwable err) {
        // なんかエラー
        Log.e(TAG, "error() called.", err);
    }
});

Thread.sleep(3000); // 3秒待って

closer.Close(); // Listen 終了

すばらしい！ .sample(1, TimeUnit.SECONDS) を挟むだけで、「１秒間隔での取得」に置き換えることができます。

他にもいろいろ便利な機能がある reactive4java ですが、とりまこんなもんで。

追記 9.27

実はこのプログラム、Cold でしたー。というわけで、Hot についての記事を書きました。
* Cold を Hot にできる。そう、Publish ならね。 - Qiita

参考

Reactive Extensionsで非同期処理を簡単に - Rx の一番わかりやすいスライドだと思います！
neue cc - Rx
Reactive Extensions - かずきのBlog@Hatena
reactive4java
TYPE_ORIENTATIONを使わずに方位角を取得 - プログラマー'sペイジ
sato-c / sensorcheck - 方位を取得する処理を移植させて頂きました