What is the suitable http content-type for consuming an asynchronous stream of data?

Question

I am new to reactive programming. And my application uses the combination of Angular and Spring Webflux.

So I have a server Endpoint that produces the stream of Json_values with some delay 1sec.[Spring - webflux]

@CrossOrigin 
@GetMapping(value = "/flux" , produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Flux<ObjType> returnFlux()
{
    return Flux.just(Obj1, Obj2, Obj3,...)
        .delayElements(Duration.ofSeconds(1));
}

And my client ie. Angular Http is expected to log the response as soon as the chunk of data is available. [Angular] my code looks like this

public getResponse()
{
    var _reqOptionsArgs= { headers: new HttpHeaders().set( 'Content-Type', 'application/json' ) };

    var observable = this.httpClient.get<any>("http://localhost:8080/flux", _reqOptionsArgs);

    observable.subscribe(val => console.log(val),
                         err => console.error(err),
                         () => console.log('Observer got a complete notification'));
}

Now I expect my method to log the json objects on console as soon as the flux emits them. But the browser throws an exception after waiting for 5 seconds when flux emits 5 values :

SyntaxError: Unexpected token { in JSON at position 8
    at JSON.parse (<anonymous>)
    at XMLHttpRequest.onLoad (http://localhost:4200/vendor.js:9848:51)
    at ZoneDelegate.invokeTask (http://localhost:4200/polyfills.js:3397:31)
    at Object.onInvokeTask (http://localhost:4200/vendor.js:71849:33)
    at ZoneDelegate.invokeTask (http://localhost:4200/polyfills.js:3396:60)
    at Zone.runTask (http://localhost:4200/polyfills.js:3174:47)
    at ZoneTask.invokeTask [as invoke] (http://localhost:4200/polyfills.js:3471:34)
    at invokeTask (http://localhost:4200/polyfills.js:4609:14)
    at XMLHttpRequest.globalZoneAwareCallback (http://localhost:4200/polyfills.js:4646:21)

But when I trigger the localhost:8080/flux from Browser, it prints the response with as soon as the the object is emitted.

Maybe I am providing an wrong content-type for the response in httpClient in angular. But at the same time, I am curious that how the browser responds to that chunk of data.

Answer 1

The particular error you're getting is because Streaming JSON (application/stream+json) and plain old JSON (application/json) are two distinct content types. Streaming JSON differs in that it's a collection of individual JSON objects delimited by another value (usually a newline), and in itself, it's therefore not valid JSON.

Having said that, you're going to struggle using HttpClient for this - it's designed to read a full response and then close, whereas you're sending it a stream of JSON instead. (In essence, it's the wrong tool for the job.)

You almost certainly want to use EventSource instead, similar to:

var es = new EventSource("http://localhost:8080/flux");
es.onmessage = function(e) {
    console.log(e.data);
}

You can then call JSON.parse() or similar on e.data, and do what you like from there.

The only downside here is that EventSource doesn't let you set HTTP headers without a polyfill - but it looks like that may not be a problem here.