标签: subscriber

谁能解释我为什么这样的代码:

 networApi.getList()
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnError(throwable -> {
                throwable.getMessage();
            })
            .doOnNext(list -> {
                coursesView.populateRecyclerView(list);
                courseList = (List<Course>) courses;
            }).subscribe();

如果没有互联网进入doOnError但是进一步抛出它以便应用程序关闭,但代码如下:

networkApi.getList()
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Subscriber<List<? extends Course>>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {
                    e.getMessage();
                }

                @Override
                public void onNext(List<? extends Course> list) {
                    coursesView.populateRecyclerView(list);
                    courseList = (List<Course>) list;
                }
            });

按照我的预期工作,这意味着当没有互联网连接时,它什么都不做.

我想运行2个异步任务,一个跟随另一个(顺序).我读过有关ZIP或Flat的内容,但我对此并不了解......

我的目的是从本地SQLite加载数据,当它完成时,它将查询调用到服务器(远程).

有人可以建议我,一种实现这一目标的方法吗？

这是我正在使用的RxJava Observable骨架(单个任务):

    // RxJava Observable
    Observable.OnSubscribe<Object> onSubscribe = subscriber -> {
        try {

            // Do the query or long task...

            subscriber.onNext(object);
            subscriber.onCompleted();
        } catch (Exception e) {
            subscriber.onError(e);
        }
    };

    // RxJava Observer
    Subscriber<Object> subscriber = new Subscriber<Object>() {
        @Override
        public void onCompleted() {
            // Handle the completion
        }

        @Override
        public void onError(Throwable e) {
            // Handle the error
        }

        @Override
        public void onNext(Object result) {

          // Handle the result

        }
    };

    Observable.create(onSubscribe)
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(subscriber);

我现在开始使用Pubsubhubbub(以及所有关于实时的东西),但我在使用Subscriber选项时遇到了麻烦.

我正在尝试用PHP开发一个webapp来:

1. 订阅RSS(以前发布)到Hub(http://pubsubhubbub.appspot.com/);
2. 从Hub读取订阅的通知(更新); 没有成功!:(

我验证在PHP中存在一个库到订阅服务器(在Git中),但使用这个库不能使订阅工作(得到409错误!).

我怎样才能做到这一点？

当用户在我的iPad报亭应用程序上购买杂志订阅时,系统会提示他们分享一些信息:

分享您的信息？[app name here]的发布者希望根据他们的隐私政策使用您的姓名,电子邮件和邮政编码.

但是,我似乎无法在iTunes Connect中找到订阅者信息.谁能帮我找到这些信息？

谢谢.

我们有SQL Server数据库设置.我们正在建立一个复制方案,我们有一个发布者和订阅者.订阅者将用作报告平台,以便我们可以运行所需的所有BI查询,并且必须访问从客户端获取所有数据的服务器.订户设置为从分发者中提取数据.

我们在发布者数据库上没有很多索引,但我们需要在报告服务器(即订阅者)上使用它们.

我的问题是:SQL Server a)是否允许这种情况,并指出订阅者的任何更改都不会被推回发布者.b)如果运行快照我假设它会覆盖我们的索引,我可以阻止这种情况发生吗？c)这是一个明智的行动方案.

谢谢.

保罗金兰

http://www.topicala.com/ http://www.thecompanything.com/

我正在尝试编写一个模块,将Magento中的新闻通讯订阅者与外部数据库同步.我需要能够以编程方式更新Magento中的订阅状态,但我正在让Mucnto中的"setStatus"方法工作.它不会抛出任何错误,但代码似乎没有任何影响.下面是我调用该方法的代码:

$collection = Mage::getResourceModel('newsletter/subscriber_collection')->showStoreInfo()->showCustomerInfo();

foreach ($collection as $cust) {
    $cust->setStatus(1);
}

从理论上讲,这应该将我所有订阅者的状态设置为"已订阅".我可以选择将发送到"setStatus"的参数更改为以下任何一个不同状态的int.

1:订阅2:状态未激活3:取消订阅

如何最好地更改订户状态或使此代码工作？

我想了解如何Subscriber和Publisher在工作的java 9.

这里我在subscriber这里创建了一个并SubmissionPublisher用于发布项目.

我正在尝试发布100个字符串subscriber.如果我不编写Client程序sleep(请参阅注释代码MyReactiveApp),我看不到所有项目都已发布.

为什么不等待这里处理的所有字符串:

strs.stream().forEach(i -> publisher.submit(i)); // what happens here? 

如果我用上面的代码替换,我看到所有字符串都在控制台中打印

strs.stream().forEach(System.out::println);

使用发布的客户端程序SubmissionPublisher.

import java.util.List;
import java.util.concurrent.SubmissionPublisher;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class MyReactiveApp {

    public static void main(String args[]) throws InterruptedException {

        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        MySubscriber subs = new MySubscriber();
        publisher.subscribe(subs);


        List<String> strs = getStrs();

        System.out.println("Publishing Items to Subscriber");
        strs.stream().forEach(i -> …

我想要进行单元测试并获得所有代码的覆盖率，但我无法获得订阅内存在的代码的覆盖率我能够监视服务和功能，但在订阅内我无法进行单元测试并获得代码覆盖率。以下是 Angular 7 代码。

LoadListData(type) {
    this.itemListEnvName = [];
    if (type === 'EnvirnmentList') {
      this.environmentBookingService.getBookedEnv()
        .subscribe(
          (environmentBookingsData: EbEnvironmentBooking[]) => {
            if (environmentBookingsData.length > 0) {
              this.itemListEnvNameList = environmentBookingsData;
              this.itemListEnvName = [];
              this.itemListEnvNameList.forEach(element => {
                const obj = {};
                obj['id'] = element['environmentId'];
                obj['itemName'] = element['environmentName'];
                this.itemListEnvName.push(obj);
                this.generateCheckDisable = false;
              });
            } else {
              this.generateCheckDisable = true;
            }
          },
          (error) => {
            this.showMessage('No Response From Delivery DB API');
          }
        );
    } else {

      this.showMessage('No Response From Delivery DB API');
    }


  }

单元测试用例内的代码就像 …

当我查看机器人操作系统（ROS）的教程时，我发现大多数示例代码将发布者的队列大小设置为较大的值，例如1000。我认为这会导致节点失去实时响应。

人们出于什么目的将其设置为那么大的值？

我在google pubsub中创建了一个主题，并在主题内创建了一个订阅，具有以下设置

然后我在 go 中编写了一个拉取器，使用它的Receive来拉取并确认已发布的消息

package main

import (
    ...
)

func main() {
    ctx := context.Background()

    client, err := pubsub.NewClient(ctx, config.C.Project)
    if err != nil {
       // do things with err
    }
    sub := client.Subscription(config.C.PubsubSubscription)
    err := sub.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
        msg.Ack()
    })

    if err != context.Canceled {
      logger.Error(fmt.Sprintf("Cancelled: %s", err.Error()))
    }
    if err != nil {
      logger.Error(fmt.Sprintf("Error: %s", err.Error()))
    }
  }

没什么特别的，它工作得很好，但过了一段时间（~闲置 3 小时后），它停止接收新发布的消息，没有错误，什么也没有。我错过了什么吗？