IntentImportProcessor { @NonNull public synchronized Maybe<IntentImportResult> process(@NonNull final Intent intent) { return Maybe.fromCallable(() -> { if (intent.hasExtra(INTENT_CONSUMED)) { return null; } else { final Uri uri; if (Intent.ACTION_VIEW.equals(intent.getAction()) && intent.getData() != null) { uri = intent.getData(); } else if (Intent.ACTION_SEND.equals(intent.getAction()) && intent.getExtras() != null) { uri = (Uri) intent.getExtras().get(Intent.EXTRA_STREAM); } else { uri = null; } if (uri != null) { return buildResultFromUri(uri); } else { return null; } } }) .doOnSuccess(intentImportResult -> { Logger.debug(IntentImportProcessor.this, "Successfully processed the file {} with uri: {}.", intentImportResult.getFileType(), intentImportResult.getUri()); analytics.record(new DefaultDataPointEvent(Events.Intents.ReceivedActionableIntent).addDataPoint(new DataPoint("type", intentImportResult.getFileType()))); lastResult.onNext(Optional.of(intentImportResult)); }) .subscribeOn(Schedulers.io()); } @Inject  IntentImportProcessor(@NonNull Context context, @NonNull Analytics analytics); @NonNull synchronized Maybe<IntentImportResult> process(@NonNull final Intent intent); synchronized void markIntentAsSuccessfullyProcessed(@NonNull Intent intent); synchronized Observable<Optional<IntentImportResult>> getLastResult();  }
@Test public void processMainIntent() { final Intent intent = new Intent(Intent.ACTION_MAIN); final TestObserver<IntentImportResult> testObserver = intentImportProcessor.process(intent).test(); testObserver.awaitTerminalEvent(); testObserver.assertNoValues() .assertComplete() .assertNoErrors(); }