标签: reactive-programming

关于 ReactiveUi 的另一个问题。我有一个用于编辑表单的 ViewModel。模型是 ReactiveObject。我只想在对象发生更改时启用 savecommand。我的尝试：

var canSaveCommand =
        this.WhenAnyValue(vm => vm.CurrentClient)
            .Where(client => client != null)
            .Select(client =>
                client.Changed
            )
            .Any();

但是当表单出现时SaveCommand已经启用了。我的错在哪里？

我们正在使用 Flatlist 开发一个 React Native 应用程序。绑定来自 API 服务的数据及其工作正常。假设服务中没有可用数据，我们需要为此显示单独的设计。

我们为此使用了“renderEmptyListComponent”

共享代码，请检查

<FlatList style={{ backgroundColor: 'white' }}
              data={this.state.dataSource}
              renderItem={({ item }) => (this.renderMovie(item))}
              keyExtractor={item => item.salesID}
              renderEmptyListComponent= {this.noItemDisplay}
              ItemSeparatorComponent={this.renderSeparator}>
            </FlatList>

请指导我如何做到这一点？

任何人都可以在这个路由示例中告诉我，每次我输入 ex: /api/person/1 等时，它都会转到 /api/person 吗？无论我选择哪种方法，它总是与 /api/person 搭配。

@Bean
    public RouterFunction<ServerResponse> monoRouterFunction(PersonService personService) {

      return  RouterFunctions
                .nest(path("/api/person"),
                        route(method(GET), personService::findAllPeople)
                            .andRoute(GET("/{id}"), personService::findOnePerson)
                            .andRoute(POST("/add"), personService::addPerson)
                            .andRoute(PUT("/update"), personService::updatePerson)
                            .andRoute(DELETE("/delete/{id}"), personService::deletePerson));

}

我有一个自动完成输入，当用户输入时，它从多个端点获取数据，例如：

//service call to fetch data and return as single observable
getAutocompleteSuggestions() {
    const subs$ = [
        this.http.get(endpoint1),
        this.http.get(endpoint2),
        this.http.get(endpoint3)
    ];

    return Observable.forkJoin(...subs$);
}

这些端点中的每一个都返回以下形式的数据：

{ data: [], status: xyz }

我想使用 switchmap，因为我只想显示最终调用的结果，并尝试了以下操作：

   this.getAutocompleteSuggestions(query)
          .switchMap(res => {
             return res.data;
           })
          .subscribe((results: any) => {
            this.results = results;
          });

但是 switchmap 中的“res”是一个数组，知道结果如何包含一个包含来自任意数量可观察对象响应数据的单个数组吗？

我有一个函数可以向服务器发出 API 请求。我想循环它直到它返回false（没有更多数据）。

func getData(id: Int) -> Observable<Bool> {
     return Observable.create { observer in
           // Alamofire request
           // parse data
           // if can decode, 
           //   return true and increment page's property 
           // otherwise false
           // error, if there's a problem
     }
}

• 第一次尝试：我试过使用takeWhile，比如 : getData(id).takeWhile {$0}。它只迭代我的函数 1x 次。

• 第二次尝试：使用范围。这里的问题是，即使我的getData函数出错，循环也不会停止，而是继续！

  Observable.range(start: 1, count: 100)
    .enumerated()
    .flatMapLatest({ _ in
          self.getData(someID)
    })
    .subscribe(onNext: { _ in
         // save to DB
         observer.onNext(true)
         observer.onCompleted()
     }, onError: …

  Run Code Online (Sandbox Code Playgroud)

当我尝试执行 Mono insidedoFinally子句时遇到问题。这是我的代码。

public interface Locks {

    Mono<ReactiveDistributedLock> doLock(LockParams params);

    Mono<Boolean> doUnlock(ReactiveDistributedLock lock);

    default <T> Mono<T> withLock(LockParams params, Supplier<Mono<T>> stage) {
        return doLock(params)
                .flatMap(lock -> stage.get().doFinally(ignored -> doUnlock(lock)));
}

问题是doUnlock(lock)insidedoFinally()返回一个没有人订阅的单声道，因为doFinally它没有链接。所以异步代码部分doUnlock从未真正执行过。

有没有办法避免这种使用Mono或Flux助手？

我正在使用流来监控我的Firestore后端的变化。这适用于更新，但我没有得到初始值。

class A {
  final Stream<DocumentSnapshot> _visitorCount =
      Firestore.instance.document('ServerData/serverStatus').snapshots();

  Stream<int> get visitorCount {
    return Observable(_visitorCount)
        .map((DocumentSnapshot doc) => doc['activeUsers'] as int);
  }
}

我可以使用rxdart'sstartWith来提供一个初始值，但是读出这个值会导致 aFuture<int>并且我需要提供一个intto startWith。

static Future<int> f = Firestore.instance
  .collection('ServerData')
  .document('serverStatus')
  .get()
  .then((doc) => doc['activeUsers'] as int);

在路上，我像这样使用它：

StreamBuilder(
          stream: _visitorCount,
          initialData: 0,
          builder: (context, snapshot) =>
              Text('${snapshot.data} users logged in') ...

我提供0初始值，但这当然只是一个临时值。我相信这很明显，但我仍然很难做到正确。解决这个问题的最佳方法是什么？

我想在Flux完成后返回一个值（或发布者）。这是一个带有（伪）代码的示例，类似于我所追求的：

val myId : Mono<String> = fetchMyId()
myId.flatMap { id ->
     someFlux.map { .. }.doOnNext { ... }.returnOnComplete(Mono.just(id))
}

即我想id在someFlux完成后返回。该returnOnComplete函数是虚构的并且不存在（有一个doOnComplete函数，但它是为了副作用）这就是我问这个问题的原因。我怎样才能做到这一点？

我目前开始使用 micronaut 和 kotlin。我有 JPA 查询，结果大约有 100 万个。这些结果我想从这个 micronaut 服务流到另一个。

我的查询返回 aallQuery.resultStream类型java.util.stream。

发送服务的控制器：

@Get("/test{value1,value2,value3}")
fun getTestObjects(
    value1: String,
    value2: String,
    value3: String
): Stream<TestObject> {
    val entries = testRepository.findAllWhere(value1, value2, value3)

    return entries
}

接收服务的客户：

@Get("/data/test{value1,value2,value3}")
override fun getTestObjects(alue1: String,
    value2: String,
    value3: String) : Stream<TestObject>

JPA 查询如下所示：

    val cb = entityManager.criteriaBuilder
    val cq = cb.createQuery(TestObject::class.java)

    val rootEntry = cq.from(TestObject::class.java)

    val predicates = mutableListOf<Predicate>()

    predicates.add(cb.like(rootEntry.get<String>("value1"), value1))
    predicates.add(cb.equal(rootEntry.get<String>("value2"), value2))
    predicates.add(cb.equal(rootEntry.get<Int>("value3"), value3))

    val cqAllWhere = cq.select(rootEntry)
        .where(cb.or(*predicates.toTypedArray()))

    val …

我有两个对象，其中每个对象都包含列表，并且我的整个 Spring Boot 项目正在使用项目反应器构造。我应该在通量或单声道映射操作中使用 java 流来遍历和操作数据吗？

Flux.fromIterable(page.getPageFieldData()).subscribeOn(Schedulers.elastic()).map(pageField - > {
        clientDataMono.subscribeOn(Schedulers.elastic()).map(clientData - > {
            Flux.fromIterable(clientData.getPageFieldData())
            .filter(clientPageField - > clientPageField.getId() == pageField.getId())
            .subscribeOn(Schedulers.elastic()).map(field - > {
                dataUpdated = true;
                pageField.setData(field.getData());
                return field;
            }).subscribe();
            return clientData;
        }).subscribe();
        if (dataUpdated) {
            pageField.setModifiedOn(Instant.now(Clock.systemUTC()));
        }
        return pageField;
    }).subscribe();

或者我应该在通量的映射操作中使用 java 流。