Java FlowableSubscriber类代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Java›Java编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Java中io.reactivex.FlowableSubscriber类的典型用法代码示例。如果您正苦于以下问题：Java FlowableSubscriber类的具体用法？Java FlowableSubscriber怎么用？Java FlowableSubscriber使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

FlowableSubscriber类属于io.reactivex包，在下文中一共展示了FlowableSubscriber类的5个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Java代码示例。

示例1: main

import io.reactivex.FlowableSubscriber; //导入依赖的package包/类
public static void main(String[] args) {
	MathFlowable.averageDouble(Flowable.range(10, 9)).subscribe(new FlowableSubscriber() {

		@Override
		public void onComplete() {
			// TODO Auto-generated method stub
			System.out.println("completed successfully");
		}

		@Override
		public void onError(Throwable arg0) {
			// TODO Auto-generated method stub

		}

		@Override
		public void onNext(Object value) {
			// TODO Auto-generated method stub
			System.out.println("average:-" + value);
		}

		@Override
		public void onSubscribe(Subscription subscription) {
			// TODO Auto-generated method stub
			subscription.request(1);
		}
	});
}

开发者ID:PacktPublishing，项目名称:Reactive-Programming-With-Java-9，代码行数:29，代码来源:Demo_average.java

示例2: flowable

import io.reactivex.FlowableSubscriber; //导入依赖的package包/类
private void flowable() {
    Flowable.create(new FlowableOnSubscribe<Integer>() {
        @Override
        public void subscribe(@NonNull FlowableEmitter<Integer> e) throws Exception {
            Log.e(TAG, "start send data ");
            for (int i = 0; i < 100; i++) {
                e.onNext(i);
            }
            e.onComplete();
        }
    }, BackpressureStrategy.DROP)//指定背压策略
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new FlowableSubscriber<Integer>() {
                @Override
                public void onSubscribe(@NonNull Subscription s) {
                    //1， onSubscribe 是2.x新添加的方法，在发射数据前被调用，相当于1.x的onStart方法
                    //2， 参数为  Subscription ，Subscription 可用于向上游请求发射多少个元素，也可用于取笑请求
                    //3,  必须要调用Subscription 的request来请求发射数据，不然上游是不会发射数据的。
                    Log.e(TAG, "onSubscribe...");
                    s.request(10);
                }

                @Override
                public void onNext(Integer integer) {
                    Log.e(TAG, "onNext:" + integer);
                }

                @Override
                public void onError(Throwable t) {
                    Log.e(TAG, "onError..." + t);
                }

                @Override
                public void onComplete() {
                    Log.e(TAG, "onComplete...");
                }
            });

}

开发者ID:wuhighway，项目名称:DailyStudy，代码行数:41，代码来源:RxJavaActivity.java

示例3: forkEach

import io.reactivex.FlowableSubscriber; //导入依赖的package包/类
/**
 * @param flowable
 * @param doWithRegistrySpec
 * @param <T>
 * @return
 * @see RxRatpack#forkEach(Observable, Action)
 */
public static <T> Flowable<T> forkEach(Flowable<T> flowable, Action<? super RegistrySpec> doWithRegistrySpec) {
  return flowable.lift(downstream -> new FlowableSubscriber<T>() {

    private final AtomicInteger wip = new AtomicInteger(1);
    private final AtomicBoolean closed = new AtomicBoolean();
    private Subscription subscription;

    @Override
    public void onSubscribe(Subscription s) {
      this.subscription = s;
      s.request(1);
      downstream.onSubscribe(s);
    }

    @Override
    public void onComplete() {
      maybeDone();
    }

    @Override
    public void onError(final Throwable e) {
      terminate(() -> downstream.onError(e));
    }

    private void maybeDone() {
      if (wip.decrementAndGet() == 0) {
        terminate(downstream::onComplete);
      }
    }

    private void terminate(Runnable runnable) {
      if (closed.compareAndSet(false, true)) {
        subscription.cancel();
        runnable.run();
      }
    }

    @Override
    public void onNext(final T t) {
      // Avoid the overhead of creating executions if downstream is no longer interested
      if (closed.get()) {
        return;
      }

      wip.incrementAndGet();
      Execution.fork()
        .register(doWithRegistrySpec)
        .onComplete(e -> this.maybeDone())
        .onError(this::onError)
        .start(e -> {
          if (!closed.get()) {
            subscription.request(1);
            downstream.onNext(t);
          }
        });
    }
  });
}

开发者ID:drmaas，项目名称:ratpack-rx2，代码行数:66，代码来源:RxRatpack.java

示例4: go

import io.reactivex.FlowableSubscriber; //导入依赖的package包/类
@OnClick(R.id.go)
public void go() {
    Editable url = urlView.getText();

    if (TextUtils.isEmpty(url)) {
        Toast.makeText(this, "url is empty", Toast.LENGTH_SHORT);
        return;
    }

    RxEasyHttp.get(url.toString(), new RxEasyStringConverter())
            .doOnSubscribe(new Consumer<Subscription>() {
                @Override
                public void accept(@NonNull Subscription subscription) throws Exception {
                    dialog.show();
                    body.setText("");
                }
            })
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new FlowableSubscriber<String>() {
                @Override
                public void onSubscribe(Subscription s) {
                    s.request(Long.MAX_VALUE);
                    dialog.show();
                    body.setText("");
                }

                @Override
                public void onNext(String response) {
                    body.setText(response);
                }

                @Override
                public void onError(Throwable t) {
                    body.setText(t.toString());
                }

                @Override
                public void onComplete() {
                    dialog.cancel();
                }
            });
}

开发者ID:LaurenceYang，项目名称:EasyHttp，代码行数:43，代码来源:RxGetActivity.java

示例5: submit

import io.reactivex.FlowableSubscriber; //导入依赖的package包/类
@OnClick(R.id.submit)
public void submit() {
    Editable content = comment.getText();

    if (TextUtils.isEmpty(content)) {
        Toast.makeText(this, "comment is empty", Toast.LENGTH_SHORT);
        return;
    }

    EasyRequestParams params = new EasyRequestParams();
    params.put("content", content.toString());

    String url = "http://book.km.com/app/index.php?c=version&a=feedback";

    RxEasyHttp.post(url, params, new RxEasyCustomConverter<PostEntity>() {
                @Override
                public void doNothing() {
                    // 防止范型类型擦除引起范型类型不能正确获取问题.
                }
            })
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new FlowableSubscriber<PostEntity>() {

                @Override
                public void onSubscribe(Subscription s) {
                    s.request(Long.MAX_VALUE);
                    dialog.show();
                }

                @Override
                public void onNext(PostEntity entity) {
                    Toast.makeText(RxPostActivity.this, "提交成功", Toast.LENGTH_LONG).show();
                    result.setText("status : " + entity.getStatus() + "\n" +
                            "message : " + entity.getMessage());

                }

                @Override
                public void onError(Throwable t) {
                    Toast.makeText(RxPostActivity.this, "提交失败", Toast.LENGTH_LONG).show();
                    result.setText(t.getMessage());
                    dialog.cancel();
                }

                @Override
                public void onComplete() {
                    dialog.cancel();
                }
            });
}

开发者ID:LaurenceYang，项目名称:EasyHttp，代码行数:51，代码来源:RxPostActivity.java

注：本文中的io.reactivex.FlowableSubscriber类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。