标签: reactive-programming

我尝试使用 RxJS 来模拟一些游戏。但我发现循环依赖存在一些问题。因此，我将游戏简化为简单的模拟（我只留下了“移动”动作）。你可以在下面找到代码（我省略了一些部分，你可以在这里找到repo）

const rx = require('rx')
const math = require('mathjs')
const _ = require('underscore')

const FIELD_SIZE = 10

const ctx = require('axel');

let getInitialState = () => {
  return {
    size: FIELD_SIZE,
    people: [
      { x: 0, y: 0 },
      { x: 9, y: 9 },
      { x: 5, y: 5 }
    ]
  }
}

var drawWorld = ({size, people}) => {
  // draw world logic
}


let getMove = (index)=> {
  let [xOffset, yOffset] = [[0,1], …

我下面有两个测试函数，我希望返回相同的结果。第一个 (monoTest3) 在 '.then()' 方法中使用 lambda 表达式。第二个没有。

为什么我会得到我所做的结果？

使用 ReactiveX 和 lambda 表达式时有哪些注意事项？

在下面的代码中，monoTest3 产生

monoVoid called
Success
----
monoVoid2 called
Success

和 monoTest4 产生（所需的结果）：

monoVoid called
monoVoid2 called
Success
----
monoVoid2 called
monoVoid called
Success

最后是代码。

private static Mono<Void> monoVoid() {
    System.out.println("monoVoid called");
    return Mono.empty();
}

private static Mono<Void> monoVoid2() {
    System.out.println("monoVoid2 called");
    return Mono.just("Hello").then();
}

@Test
public void monoTest3() throws Exception {
    monoVoid()
            .then( v -> monoVoid2())
            .doOnSuccess(v -> System.out.println("Success"))
            .block();

    System.out.println("----");

    monoVoid2()
            .then( v -> monoVoid())
            .doOnSuccess(v …

当我收到某个值时，如何取消订阅可观察值？

像这样 ：

 let tempSub: Subscription = this.observable$.subscribe(value => {
    if (value === 'somethingSpecific') {
      tempSub.unsubscribe(); 
      // doesn't work 
      //because when this is reached tempsub is undefined
    }
  });

我几乎是 ReactiveCocoa/ReactiveSwift 的菜鸟。我有两个 SignalProducers。如果第一个 SignalProducer 返回 nil，那么我想执行第二个 SignalProducer，否则不执行。我阅读了文档，但我不确定哪种语法可以帮助我完成这样的工作。任何帮助都受到高度赞赏。

我有多个热的 observables，它们可能会或可能不会发出项目。因此，我想组合 observables，然后在它们中的任何一个发出结果时处理结果，但如果其他 observables 在 item 处发出，则它们应该一起处理。

例如。

observable1 = PublishSubject<>()  
observable2 = PublishSubject<>()

observable1.onNext(1)  
observable1.onNext(2)  
observable2.onNext("Test")  
observable1.onNext(3)

应该发出：

(1, null) 
(2, null)
(2, "Test")
(3, "Test")

也有可能observable2在之前被发射observable1

CombineLatest是最接近我需要的，但只有在所有可观察对象至少发出一项时才会发出结果。是否有一个反应式运算符？

哪些情况会导致Flux::flatMap同时收听多个源（0...无穷大）？

我在实验时发现，当上游向flatMap线程内发送信号时thread-upstream-1，并且存在NflatMap 将侦听的内部流，并且每个内部流都在不同的线程中发送信号：thread-inner-stream-ifor 1<=i<=N，而不是每个1<=i<=Nif thread-upstream-1 != thread-inner-stream-i，flatMap将同时侦听所有内部流溪流。

我认为这并不完全正确，我错过了一些其他场景。

我正在学习反应堆核心并关注这个https://www.baeldung.com/reactor-core

ArrayList<Integer> arrList = new ArrayList<Integer>();
System.out.println("Before: " + arrList);
Flux.just(1, 2, 3, 4)
  .log()
  .map(i -> i * 2)
  .subscribeOn(Schedulers.parallel())
  .subscribe(arrList::add);

System.out.println("After: " + arrList);

当我执行上面的代码行时，给出。

 Before: []
 [DEBUG] (main) Using Console logging
 After: []

上面的代码行应该在另一个线程中开始执行，但它根本不起作用。有人可以帮我吗？

我有一个带有方法 context() 的类 PublishContext 如下：

public static Mono<Object> context(){

    return Mono.empty().subscriberContext( context -> {
        Context context1 = context.put("key", "hello");
        System.out.println((String) context1.get("key"));
        return context1;
    });
}

在上面的代码中，上下文对象是 Context0@744 和 context1 是 Context@747 这是可以理解的，因为上下文是不可变的并且总是返回一个新的上下文。

在我的主课中，我有以下代码：

public static void main(String[] args) {

    Mono<Object> objectMono = PublishContext.context();
    objectMono.subscribe();

    Mono<Object> objectMono1 = Mono.subscriberContext().flatMap(context ->       {
        System.out.println((String) context.get("key"));
        return Mono.empty();
    });
    objectMono1.subscribe();
}

在这里，我得到的上下文为 Context0@744，即旧上下文，因此得到“上下文为空”异常。有人可以解释这种行为吗？另外，如何访问我从 context() 方法返回的 context0@747？

我在尝试缓存Mono由WebClient. 代码是这样的：

public Mono<Token> authenticate() {
    return cachedTokenMono = cachedTokenMono
        .switchIfEmpty(
            Mono.defer(() -> 
                    getToken())
                    .cache(token ->
                               Duration.between(Instant.now(), token.getExpires().toInstant()),
                           (Throwable throwable) -> Duration.ZERO,
                           () -> Duration.ZERO));
}

目的是Mono用于接收 a 的Token被缓存，直到令牌过期。令牌过期后，缓存Mono变为空并请求新令牌。这按预期工作，但不幸的switchIfEmpty()是实际上并没有“切换”，而是包装了源代码Mono。结果，随着越来越多的包装SwitchIfEmptyMono被创建，这会造成内存泄漏。在这种情况下，正确的模式是什么？有没有办法Mono用新的代替空的？

我有一个发布可选输出类型的发布者。flatMap如果输出不是nil，我需要到一个新的发布者，如果是，则回退到一个空的发布者nil。

例如，类似于：

[1, nil, 5].publisher // Generic parameter 'T' could not be inferred
    .flatMap {
        $0?.someNewPublisher ?? Empty(completeImmediately: false)
    }

[1, nil, 5].publisher
    .map {
        $0?.someNewPublisher
    }
    .replaceNil(with: Empty(completeImmediately: false)) // Generic parameter 'Failure' could not be inferred
    .flatMap { $0 }

我想知道我是否试图以错误的方式解决这个问题。需要明确的是，nil在映射之前过滤并不能解决我的问题，因为这不会用一个空的发布者替换当前的发布者（我将继续接收我不应该再接收的元素）。