Scan_Bluetooth_With_RxJava [In Android]

public class RxBLE2 {

    private final BluetoothAdapter bluetoothAdapter;
    private final BluetoothLeScanner scanner;

    private Subscriber<? super BleDevice> subscriber;
    private final ScanCallback callback = new ScanCallback() {
        @Override
        public void onScanResult(int callbackType, ScanResult result) {
            super.onScanResult(callbackType, result);
            subscriber.onNext(BleDevice.create(result));
        }
    };


    public RxBLE2(Activity activity) {
        BluetoothManager manager = (BluetoothManager) activity.getSystemService(Context.BLUETOOTH_SERVICE);
        bluetoothAdapter = manager.getAdapter();
        scanner = bluetoothAdapter.getBluetoothLeScanner();
    }


    public Observable<BleDevice> observe() {
        if (!BluetoothHelper.IS_BLE_SUPPORTED || !bluetoothAdapter.isEnabled()) {
            return Observable.empty();
        }

        return Observable.create(new Observable.OnSubscribe<BleDevice>() {
            @Override
            public void call(Subscriber<? super BleDevice> subscriber) {
                RxBLE2.this.subscriber = subscriber;
                scanner.startScan(callback);
            }
        }).distinctUntilChanged().doOnSubscribe(() -> {
            if (bluetoothAdapter != null && bluetoothAdapter.isEnabled()) {
                scanner.stopScan(callback);
            }
        });
    }
}