Spring Boot 2 Reactor Flux API的Angular客户端

Question

如何为Java Project Reactor反应FluxAPI 创建Angular 4客户端？下面的示例有两个API:MonoAPI; 和,FluxAPI.两者都起作用curl; 但在Angular 4(4.1.2)中,只有MonoAPI有效; 任何想法如何让Angular 4与FluxAPI 一起工作？

这是一个简单的Spring Boot 2.0.0-SNAPSHOT应用程序,它带有MonoAPI和FluxAPI:

@SpringBootApplication
@RestController
public class ReactiveServiceApplication {

    @CrossOrigin
    @GetMapping("/events/{id}")
    public Mono<Event> eventById(@PathVariable long id) {
        return Mono.just(new Event(id, LocalDate.now()));
    }

    @CrossOrigin
    @GetMapping(value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<Event> events() {
        Flux<Event> eventFlux = Flux.fromStream(
            Stream.generate(
                ()->new Event(System.currentTimeMillis(), LocalDate.now()))
            );

        Flux<Long> durationFlux = Flux.interval(Duration.ofSeconds(1));

        return Flux.zip(eventFlux, durationFlux).map(Tuple2::getT1);
    }

    public static void main(String[] args) {
        SpringApplication.run(ReactiveServiceApplication.class);
    }
}

使用Lombok编辑的活动:

@Data
@AllArgsConstructor
public class Event {
    private final long id;
    private final LocalDate when;
}

正如我所期望的那样,这些反应性API在curl中起作用:

jan@linux-6o1s:~/src> curl -s http://localhost:8080/events/123
{"id":123,"when":{"year":2017,"month":"MAY","monthValue":5,"dayOfMonth":15,"dayOfWeek":"MONDAY","era":"CE","dayOfYear":135,"leapYear":false,"chronology":{"calendarType":"iso8601","id":"ISO"}}}

类似地,对于非终止的Flux API:

jan@linux-6o1s:~/src> curl -s http://localhost:8080/events
data:{"id":1494887783347,"when":{"year":2017,"month":"MAY","monthValue":5,"dayOfMonth":15,"dayOfWeek":"MONDAY","era":"CE","dayOfYear":135,"leapYear":false,"chronology":{"calendarType":"iso8601","id":"ISO"}}}

data:{"id":1494887784348,"when":{"year":2017,"month":"MAY","monthValue":5,"dayOfMonth":15,"dayOfWeek":"MONDAY","era":"CE","dayOfYear":135,"leapYear":false,"chronology":{"calendarType":"iso8601","id":"ISO"}}}

data:{"id":1494887785347,"when":{"year":2017,"month":"MAY","monthValue":5,"dayOfMonth":15,"dayOfWeek":"MONDAY","era":"CE","dayOfYear":135,"leapYear":false,"chronology":{"calendarType":"iso8601","id":"ISO"}}}

...

使用RxJS的同样微不足道的Angular 4客户端:

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent  implements OnInit, OnDestroy {
  title = 'app works!';
  event: Observable<Event>;
  subscription: Subscription;

  constructor(
    private _http: Http
    ) {
  }

  ngOnInit() {
    this.subscription = this._http
      .get("http://localhost:8080/events/322")
      .map(response => response.json())
      .subscribe(
        e => { 
          this.event = e;
        }
      );
  }

  ngOnDestroy() {
    this.subscription.unsubscribe();
  }
}

适用于MonoAPI:

"http://localhost:8080/events/322"

但FluxAPI:

"http://localhost:8080/events"

从不触发事件处理程序,不像curl.

Answer 1

这是西蒙在他的回答中描述的一个有效的 Angular 4 SSE 示例。这需要一段时间才能拼凑起来，所以也许对其他人有用。这里的关键部分是 Zone——没有 Zone，SSE 更新将不会触发 Angular 的更改检测。

import { Component, NgZone, OnInit, OnDestroy } from '@angular/core';
import { Http } from '@angular/http';
import { Observable } from 'rxjs/Observable';
import { BehaviorSubject } from 'rxjs/BehaviorSubject';
import { Subscription } from 'rxjs/Subscription';
import 'rxjs/add/operator/map';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent  implements OnInit {
  event: Observable<MyEvent>;
  private _eventSource: EventSource;
  private _events: BehaviorSubject<MyEvent> = new BehaviorSubject<MyEvent>(null);
  constructor(private _http: Http, private _zone: NgZone) {}
  ngOnInit() {
    this._eventSource = this.createEventSource();
    this.event = this.createEventObservable();
  }

  private createEventObservable(): Observable<MyEvent> {
    return this._events.asObservable();
  }

  private createEventSource(): EventSource {
      const eventSource = new EventSource('http://localhost:8080/events');
      eventSource.onmessage = sse => {
        const event: MyEvent = new MyEvent(JSON.parse(sse.data));
        this._zone.run(()=>this._events.next(event));
      };
      eventSource.onerror = err => this._events.error(err);
      return eventSource;
  }
}

相应的 HTML 很简单：

<b>Observable of sse</b>
<div *ngIf="(event | async); let evt; else loading">
  <div>ID: {{evt.id}} </div>
</div>
<ng-template #loading>Waiting...</ng-template>

事件是微不足道的：

export class MyEvent {
  id: number;
  when: any;

  constructor(jsonData) {
    Object.assign(this, jsonData);
  }
}

由于我的 TS 不包括EventSource或Callback，我将它们存入：

interface Callback { (data: any): void; }

declare class EventSource {
    onmessage: Callback;
    onerror: Callback;
    addEventListener(event: string, cb: Callback): void;
    constructor(name: string);
    close: () => void;
}