import rx.util.functions.Action1; //导入依赖的package包/类
public void startDownload() throws DownloadDataInvalidException {
    if (TextUtils.isEmpty(filename) || TextUtils.isEmpty(url))
        throw new DownloadDataInvalidException("filename and/or url empty");

    downloadThread = Downloader.newInstance(url, filename);
    downloadThread.loadInBackground();

    // Subscribe to the progress observable
    // Sample the stream every 30 milliseconds (ignore events in between)
    // Upon receiving an event, update the views
    downloadThread.getProgressObservable()
            .sample(30, TimeUnit.MILLISECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<DownloadProgressEvent>() {
                @Override
                public void call(DownloadProgressEvent event) {
                    if (downloadThread.isKilled()) {
                        iDownloadView.switchToDownloadView();
                    } else {
                        iDownloadView.showProgress(event);
                    }
                }
            });
    iDownloadView.switchToPauseView();
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
@Test
public void observableHandlesParams() throws Exception
{
  ArgumentCaptor<Request> requestCaptor = ArgumentCaptor.forClass(Request.class);
  when(mockClient.execute(requestCaptor.capture())) //
      .thenReturn(new Response(200, "OK", NO_HEADERS, new TypedString("hello")));
  ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
  Action1<Response> action = mock(Action1.class);
  example.observable("X", "Y").subscribe(action);

  Request request = requestCaptor.getValue();
  assertThat(request.getUrl()).contains("/X/Y");

  verify(action).call(responseCaptor.capture());
  Response response = responseCaptor.getValue();
  assertThat(response.getStatus()).isEqualTo(200);
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
@Override
public void onOpen(Connection connection) {
	this.connection = connection;
	this.subscription = stream.subscribe(new Action1<Object>() {
		@Override
		public void call(Object obj) {
			onMessage(obj.toString());
		}
	});
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
public DirectedConnection(Observable<WampMessage<TMessage>> incomingMessages,
                          Observer<WampMessage<TMessage>> outgoingMessages) {
    this.incomingMessages = incomingMessages;

    this.incomingMessages.subscribe(new Action1<WampMessage<TMessage>>() {
        @Override
        public void call(WampMessage<TMessage> message) {
            messageArrived.raiseEvent
                    (this,
                     new WampMessageArrivedEventArgs<TMessage>(message));
        }
    });

    this.outgoingMessages = outgoingMessages;
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
public static void main(String[] args) {
    try {
        DefaultWampChannelFactory factory =
                new DefaultWampChannelFactory();

        WampChannel channel =
                factory.createJsonChannel(new URI("ws://127.0.0.1:8080/ws"), "realm1");

        CompletionStage open = channel.open();

        open.toCompletableFuture().get(5000, TimeUnit.MILLISECONDS);

        WampRealmServiceProvider services = channel.getRealmProxy().getServices();

        Subject<Integer, Integer> subject =
                services.getSubject(Integer.class, "com.myapp.topic1");

        Subscription disposable = subject.subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer counter) {
                System.out.println("Got " + counter);
            }
        });

        System.in.read();

        disposable.unsubscribe();

        System.in.read();

    } catch (Exception ex) {
        // Catch everything! :(
        System.out.println(ex);
    }
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
public static void hello(String... names) {
    Observable.from(names).subscribe(new Action1<String>() {
        @Override
        public void call(String s) {
            System.out.println("Hello " + s + "!");
        }
    });
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
public static void main(String[] args) {
    BankAccount myBankAccount = new BankAccount();
    
    Observable.zip(myBankAccount.getBalance(),
                   getCatalog("Montgomery Ward")
                       .mapMany(
                            new Func1<Catalog, Observable<? extends Jeans>>() {
                                @Override
                                public Observable<? extends Jeans> call(Catalog catalog) {
                                    return catalog.findJeans("38W", "38L", "blue");
                                }
                            }
                        ),
                   new Func2<Double, Jeans, Boolean>() {
                       @Override
                       public Boolean call(Double cash, Jeans jeans) {
                           if (cash > jeans.getPurchasePrice()) {
                               return jeans.purchase();
                           }
                           return Boolean.FALSE;
                       }
                   })
                   .subscribe(
                       new Action1<Boolean>() {
                           @Override
                           public void call(Boolean success) {
                               System.out.println("We purchased the jeans: " + success);
                           }
                       },
                       new Action1<Throwable>() {
                           @Override
                           public void call(Throwable t1) {
                               System.out.println(t1);
                           }
                       }
                   );
    
    pool.shutdown();
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
private void loadFromConfig()
{
	final HashMap<Class, EventListener> eventListenerHashMap = this.eventListeners;

	ConfigurationManager.getSharedManager().getMonitoredEventClasses().subscribe(new Action1()
	{
		public void call(Object o)
		{
			Class<Event> eventClass = (Class<Event>)o;
			eventListenerHashMap.put(eventClass, ListenerManager.listenerForEventClass(eventClass));
		}
	});
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
@Test
public void observableCallsOnNext() throws Exception
{
  when(mockClient.execute(any(Request.class))) //
      .thenReturn(new Response(200, "OK", NO_HEADERS, new TypedString("hello")));
  Action1<String> action = mock(Action1.class);
  example.observable("Howdy").subscribe(action);
  verify(action).call(eq("hello"));
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
@Test
public void observableCallsOnError() throws Exception
{
  when(mockClient.execute(any(Request.class))) //
      .thenReturn(new Response(300, "FAIL", NO_HEADERS, new TypedString("bummer")));
  Action1<String> onSuccess = mock(Action1.class);
  Action1<Throwable> onError = mock(Action1.class);
  example.observable("Howdy").subscribe(onSuccess, onError);
  verifyZeroInteractions(onSuccess);
  verify(onError).call(isA(RetrofitError.class));
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
@Test
public void observableUsesHttpExecutor() throws IOException
{
  Response response = new Response(200, "OK", NO_HEADERS, new TypedString("hello"));
  when(mockClient.execute(any(Request.class))).thenReturn(response);

  example.observable("Howdy").subscribe(mock(Action1.class));

  verify(mockRequestExecutor, atLeastOnce()).execute(any(Runnable.class));
  verifyZeroInteractions(mockCallbackExecutor);
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
protected void onNotifyRequested() {
    final NotifyAuthorization.Builder auth = new NotifyAuthorization.Builder();

    if (mIsEmailAuthRequired) {
        Account acc = (Account) mSpinner.getSelectedItem();
        if (acc != null) {
            // Set the selected email as the user preference
            String emailPrefKey = getActivity().getString(R.string.gmail_account_preference);
            mSharedPreferences.edit().putString(emailPrefKey, acc.name).apply();

            AccountUtils.getPreferedGmailAuth(getActivity(), mAccountManager, mSharedPreferences, getActivity()).
                    subscribeOn(Schedulers.newThread()).
                    observeOn(AndroidSchedulers.mainThread()).subscribe(new Action1<EmailAuthorization>() {
                @Override
                public void call(EmailAuthorization emailAuth) {
                    Log.d(TAG, "call() - got EmailAuthorization: " + emailAuth.getEmailAddress() + ":" + emailAuth.getToken());
                    auth.withAuth(emailAuth);
                    mGroup.setMessage(mMsgField.getText().toString().trim());
                    mDb.update(mGroup);
                    executeNotifyDraw(auth.build(), mGroup, mMemberIds);
                }
            });
        }  // else no email auth available - do nothing.
    } else {
        // We have no additional authorization - just send as is
        // Get the custom message.
        mGroup.setMessage(mMsgField.getText().toString().trim());
        mDb.update(mGroup);
        executeNotifyDraw(auth.build(), mGroup, mMemberIds);
    }
    this.dismiss();
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
@Override
protected AbstractLayout getAdvancedSearchLayout()
{
	final VerticalLayout advancedSearchLayout = new VerticalLayout();
	advancedSearchLayout.setSpacing(true);

	final HorizontalLayout tagSearchLayout = new HorizontalLayout();
	this.tagSearchField = new TagField("Search Tags", true);
	tagSearchLayout.addComponent(this.tagSearchField);

	tagSearchLayout.setSizeFull();
	advancedSearchLayout.addComponent(tagSearchLayout);

	final HorizontalLayout stringSearchLayout = new HorizontalLayout();
	stringSearchLayout.addComponent(this.searchField);
	stringSearchLayout.setWidth("100%");

	advancedSearchLayout.addComponent(stringSearchLayout);

	final Button searchButton = new Button("Search");
	final Action1<ClickEvent> searchClickAction = new SearchClickAction();
	ButtonEventSource.fromActionOf(searchButton).subscribe(searchClickAction);

	advancedSearchLayout.addComponent(searchButton);
	advancedSearchLayout.setComponentAlignment(searchButton, Alignment.MIDDLE_RIGHT);

	return advancedSearchLayout;

}
 

import rx.util.functions.Action1; //导入依赖的package包/类
@Test
public void observableApiIsCalledWithDelay()
{
  mockRestAdapter.setDelay(100);
  mockRestAdapter.setVariancePercentage(0);
  mockRestAdapter.setErrorPercentage(0);

  final Object expected = new Object();
  class MockObservableExample implements ObservableExample
  {
    @Override
    public Observable<Object> doStuff()
    {
      return Observable.from(expected);
    }
  }

  ObservableExample mockService =
      mockRestAdapter.create(ObservableExample.class, new MockObservableExample());

  final long startNanos = System.nanoTime();
  final AtomicLong tookMs = new AtomicLong();
  final AtomicReference<Object> actual = new AtomicReference<Object>();
  Action1<Object> onSuccess = new Action1<Object>()
  {
    @Override
    public void call(Object o)
    {
      tookMs.set(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos));
      actual.set(o);
    }
  };
  Action1<Throwable> onError = new Action1<Throwable>()
  {
    @Override
    public void call(Throwable throwable)
    {
      throw new AssertionError();
    }
  };

  mockService.doStuff().subscribe(onSuccess, onError);

  verify(httpExecutor, atLeastOnce()).execute(any(Runnable.class));
  verifyZeroInteractions(callbackExecutor);

  assertThat(actual.get()).isNotNull().isSameAs(expected);
  assertThat(tookMs.get()).isGreaterThanOrEqualTo(100);
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
@Override
public void onActivityCreated(Bundle savedInstanceState) {
    super.onActivityCreated(savedInstanceState);

    // Populate those field that require injected dependencies
    long groupId = getArguments().getLong(Intents.GROUP_ID_INTENT_EXTRA);
    mGroup = mDb.queryById(groupId, Group.class);
    String message = mSavedMsg == null ? mGroup.getMessage() :
            savedInstanceState.getString(MESSAGE_KEY);

    mMsgField.append(message);
    int remainingChars = mMaxMsgLength;
    if (message != null) {
        remainingChars = message.length() >= mMaxMsgLength ? 0 : mMaxMsgLength - message.length();
    }
    setCharactersRemaining(remainingChars);

    mIsEmailAuthRequired = NotifyUtils.containsEmailSendableEntry(mDb, mMemberIds);

    if (mIsEmailAuthRequired) {
        // Add all Gmail accounts to list
        final Observable<Account[]> accountsObservable = AccountUtils.getAllGmailAccountsObservable(getActivity(), mAccountManager);
        accountsObservable.
                subscribeOn(Schedulers.newThread()).
                observeOn(AndroidSchedulers.mainThread()).
                subscribe(new Action1<Account[]>() {
                              @Override
                              public void call(Account[] accounts) {
                                  AccountAdapter aa = new AccountAdapter(getActivity(), accounts);
                                  mSpinner.setAdapter(aa);
                                  mEmailFromContainer.setVisibility(View.VISIBLE);
                                  // TODO Set email to preference
                              }
                          },
                        new Action1<Throwable>() {
                            @Override
                            public void call(Throwable throwable) {
                                mInfoTextView.setText(throwable.getMessage());
                                mInfoTextView.setVisibility(View.VISIBLE);
                            }
                        }
                );
    }
}
 

import rx.util.functions.Action1; //导入依赖的package包/类
private TargetLine insertTargetLine(final int row)
{
	final List<EmailAddressType> targetTypes = getTargetTypes();

	EmailForm.this.grid.insertRow(row);
	this.grid.setCursorY(row);
	this.grid.setCursorX(0);

	final TargetLine line = new TargetLine();
	line.row = row;

	line.targetTypeCombo = new ComboBox(null, targetTypes);
	line.targetTypeCombo.setWidth("100");
	line.targetTypeCombo.select(targetTypes.get(0));
	this.grid.addComponent(line.targetTypeCombo);

	line.targetAddress = new ComboBox(null);
	this.grid.addComponent(line.targetAddress);
	line.targetAddress.setImmediate(true);
	line.targetAddress.setTextInputAllowed(true);
	line.targetAddress.setInputPrompt("Enter Contact Name or email address");
	line.targetAddress.setWidth("100%");
	line.targetAddress.addValidator(new EmailValidator("Please enter a valid email address."));

	line.targetAddress.setContainerDataSource(getValidEmailContacts());
	line.targetAddress.setItemCaptionPropertyId("namedemail");
	line.targetAddress.setNewItemsAllowed(true);

	line.targetAddress.setNewItemHandler(new NewItemHandler()
	{
		private static final long serialVersionUID = 1L;

		@Override
		public void addNewItem(final String newItemCaption)
		{
			final IndexedContainer container = (IndexedContainer) line.targetAddress.getContainerDataSource();

			final Item item = addItem(container, "", newItemCaption);
			if (item != null)
			{
				line.targetAddress.addItem(item.getItemProperty("id").getValue());
				line.targetAddress.setValue(item.getItemProperty("id").getValue());
			}
		}
	});

	line.minusButton = new Button("-");
	line.minusButton.setDescription("Click to remove this email address line.");
	line.minusButton.setData(line);
	line.minusButton.setStyleName(Reindeer.BUTTON_SMALL);
	this.grid.addComponent(line.minusButton);
	final Action1<ClickEvent> minusClickAction = new MinusClickAction();

	line.buttonSubscription = ButtonEventSource.fromActionOf(line.minusButton).subscribe(minusClickAction);

	this.lines.add(line);

	return line;
}