Angular client of Spring Boot 2 Reactor Flux API

Question

How do I create an Angular 4 client for a Java Project Reactor reactive Flux API? The sample below has two APIs: a Mono API; and, Flux API. Both work from curl; but in Angular 4 (4.1.2) only the Mono API works; any ideas how to get Angular 4 to work with the Flux API?

Here's a trivial Spring Boot 2.0.0-SNAPSHOT application with a Mono API and a 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);
    }
}

with a Lombok-ed event:

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

These reactive APIs work from curl as I'd expect:

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

and similarly for the non-terminating 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"}}}

...

The similarly trivial Angular 4 client with RxJS:

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

works fine for the Mono API:

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

but the Flux API:

"http://localhost:8080/events"

never triggers the event handler, unlike curl.

Fruit

I'm happy to clarify, especially to the vote-to-close voter because it's "unclear what you're asking" -- help me understand what's unclear in a comment? – Fruit

No one among the solutions is nice. What we need is a new version of Angular! – Edris

Answer 1

Here's a working Angular 4 SSE example as Simon describes in his answer. This took a while to piece together so perhaps it'll be useful to others. The key piece here is Zone -- without Zone, the SSE updates won't trigger Angular's change detection.

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

The corresponding HTML is simply:

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

The event is trivial:

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

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

and since my TS does not include EventSource or Callback, I stubbed them in:

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

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

Fruit

Answer 2

The Flux based controller is producing Server Sent Events (SSE). I don't think the Http client from Angular2 lets you consume SSE...

edit: looks like EventSource is what you need, see this similar question/answer: https://mcmap.net/q/890998/-using-rxjs-and-angular-2-in-order-to-deal-with-server-sent-events

Paperweight

Ahh, of course! The data: in the curl output indicates that this is a SSE text stream. Thanks, Simon! – Fruit

Thanks, again, Simon, for the pointers -- see my answer below for the details. Feel free to copy my answer into yours, and I'll accept your answer... – Fruit

Answer 3

Going to guess here that the url for /events is the problem because it should produce json to be handled.

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

Connell