Spring Boot 2 Reactor Flux API의 각 클라이언트

Question

Java Project ReactorFlux API 용 Angular 4 클라이언트는 어떻게 만듭니 까? 아래 샘플에는 두 개의 API가 있습니다. Mono API; 및 Flux API 둘 다 curl에서 작동합니다. Angular 4 (4.1.2)에서는 Mono API 만 작동합니다. Angular 4를 Flux API와 함께 사용하는 방법에 대한 아이디어가 있습니까? 롬복 에드 이벤트와

@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); 
    } 
} 

:

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

이 반응 API를에서 작업 여기

는 Mono API와 사소한 봄 부팅 2.0.0-SNAPSHOT 응용 프로그램과 Flux API입니다 나는 예상대로 말아 올릴 것이다 :

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"}}} 

그리고 non-termi 이 없습 플럭스 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과의 유사 사소한 각도 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(); 
    } 
} 

가 Mono API에 대해 잘 작동 :

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

하지만 Flux API :

"http://localhost:8080/events" 

curl과 달리 이벤트 처리기를 트리거하지 않습니다.

Answer 1

가 작동 각도 4 SSE의 예입니다 : EventSource처럼 보이는 것은 당신이 필요로하는 무슨이 비슷한 질문/답을 볼 수있다 사이먼이 그의 대답에서 묘사 하듯이. 이것은 함께 조각하기까지 시간이 걸렸으므로 아마도 다른 사람들에게 유용 할 것입니다. 핵심 요소는 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; 
} 

Answer 2

/events의 URL은 처리 할 json을 생성해야하기 때문에 문제가 될 것입니다.

@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.APPLICATION_JSON_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); 
    } 
} 

Answer 3

Flux 기반 컨트롤러는 SSE (Server Sent Events)를 생성합니다. 내가

편집 ... Angular2에서 Http 클라이언트는 SSE를 소비 할 수 있습니다 생각하지 않는다 : https://stackoverflow.com/a/36815231/1113486 여기