最佳媒体类型，可通过服务器发送事件将 Flux 提供给 Angular/RxJs/observer 尽快返回 Flux

Question

上下文：Spring Web Flux以服务器已发送事件样式将Flux返回到Angular。

个人知识：

根据其他stackoverflow主题

媒体类型applicationjson将在内存中缓冲流量，并在一次传递中将其序列化。

媒体类型applicationstream json将在网络上刷新流量输入的每个元素。当流是无限的，或者当您想在信息可用时立即将信息推送到客户端时，这种行为非常方便。

基于前两个前提，我选择了applicationstream json，因为我希望“在信息可用时尽快将其推送到客户端”，而不是“在内存中缓冲流量并在一次传递中序列化”。

好吧，如果我添加MediaType.APPLICATION_STREAM_JSON，就像我在WebFluxendpoint中一样，那么从Spring Webclient读取时就可以了，但是它不适用于通过SSE连接的Angular / RxJs。

// Delay 0.5 second
@GetMapping(path = "/search-with-delay/{parte_da_palavra}")
public Flux<Sugestao> getSugestoesDelay(@PathVariable("parte_da_palavra") String parte_da_palavra) {
    return sugestaoService.findAllMySugestoes("name", parte_da_palavra).delayElements(Duration.ofMillis(500));
}

@GetMapping(path = "/search/{parte_da_palavra}")
public Flux<Sugestao> getSugestoes(@PathVariable("parte_da_palavra") String parte_da_palavra) {
    return sugestaoService.findAllMySugestoes("name", parte_da_palavra);
}

app.component.ts

import { Component, OnInit } from '@angular/core';

//SERVICES
import { SseService } from './sse.service';
import { NosseService } from './nosse.service';

//MODELS
import { Sugestao } from './sugestao';

//RXJS
import { Observable } from 'rxjs';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent implements OnInit {
  sugestoes$: Observable<any>;
  restItems: any;

  //### SSE
  searchSseWithDelay(searchValue: string): void {
    this.sugestoes$ = this.sseService
      .getServerSentEvent("http://localhost:8080/sugestao/search-with-delay/" + searchValue);
  }

  searchSseWithoutDelay(searchValue: string): void {
    this.sugestoes$ = this.sseService
      .getServerSentEvent("http://localhost:8080/sugestao/search/" + searchValue);
  }

  //### Without SSE
  searchWithDelay(searchValue: string): void {
    this.sugestoes$ = this.nosseService.getWithoutServerSentEvent("http://localhost:8080/sugestao/search-with-delay/" + searchValue);
  }

  searchWthoutDelay(searchValue: string): void {
    this.sugestoes$ = this.nosseService.getWithoutServerSentEvent("http://localhost:8080/sugestao/search/" + searchValue);
  }

  constructor(
    private sseService: SseService, private nosseService: NosseService) { }

  ngOnInit() {
  }


}

及其服务

无上证

import { Injectable, NgZone } from '@angular/core';
import { Observable } from 'rxjs';
import { Sugestao } from './sugestao';
import { HttpClient } from '@angular/common/http';
import { map, tap } from 'rxjs/operators';

@Injectable({
    providedIn: "root"
})
export class NosseService {

    constructor(private http: HttpClient) { }

    getWithoutServerSentEvent(url: string): Observable<any> {

        return this.http
            .get<Sugestao[]>(url)
            .pipe()
            ;
    }
}

与上交所合作

import { Injectable, NgZone } from '@angular/core';
import { Observable } from 'rxjs';
import { Sugestao } from './sugestao';


@Injectable({
  providedIn: "root"
})
export class SseService {
  sugestoes: Sugestao[] = [];
  constructor(private _zone: NgZone) { }

  getServerSentEvent(url: string): Observable<any> {
    this.sugestoes = [];
    return Observable.create(observer => {
      const eventSource = this.getEventSource(url);
      eventSource.onmessage = event => {
        this._zone.run(() => {
          let json = JSON.parse(event.data);
          this.sugestoes.push(new Sugestao(json['id'], json['name'], json['phone'], json['account']));
          observer.next(this.sugestoes);
        });
      };
      eventSource.onerror = (error) => {
        if (eventSource.readyState === 0) {
          console.log('The stream has been closed by the server.');
          eventSource.close();
          observer.complete();
        } else {
          observer.error('EventSource error: ' + error);
        }
      }

    });
  }
  private getEventSource(url: string): EventSource {
    return new EventSource(url);
  }


}

谷歌搜索我发现基本上说我必须使用文本/事件流

“如果我们在不使用 Accept 标头的情况下请求内容，或者将其设置为 application/json，我们将得到一个同步的 JSON 格式响应。如果我们想在Spring中使用Server-Sent Events支持来实现我们的完整反应式堆栈，我们会在请求中（显式或在幕后）将Accept标头设置为文本/事件流，从而激活Spring中的反应式功能。

最后阅读我在Mozila文档中发现的HTML5

EventSource实例打开到HTTP服务器的持久连接，HTTP服务器以文本/事件流格式发送事件。该连接保持打开，直到调用EventSource.close（）关闭

在那之后，我真的很困惑。我肯定会使用Angular/RxJs/SSE，我的意思是，我根本不会在我的客户端使用Spring WebClient。据我所知，MediaType.APPLICATION_STREAM_JSON是我业务需求中的最佳选择，但似乎因为我使用SSE，我必须更喜欢MediaType.TEXT_EVENT_STREAM当尝试使用任何返回事件流的MediaType时，我习惯于编写观察者的方式是不正确的。

所以我直截了当的问题是：什么是最合适的MediaType设置在我的endpoint打算返回Flux很快数据可用的Angular/RxJs/Oberver使用超文本标记语言5服务器发送事件功能？

Answer 1

Spring开发者表示：

“应用程序/流json”已被弃用，取而代之的是“应用程序/x-ndjson”。杰克逊编码器和解码器现在也支持后者。虽然它不是一种官方注册的媒体类型，但至少它是一种似乎正在使用的常见约定，而JSON lines甚至没有这种约定。还有JSON文本序列，可以使用still和换行符作为分隔符，但是“application/json-seq”需要向Jackson编码器和解码器注册。至于换行符以外的分隔符，我不确定Jackson的非阻塞解析器是否支持。

总之，使用行分隔的 JSON，其中“application/x-ndjson”支持作为开箱即用的媒体类型，并用作“application/stream json”的替代品。或者，注册任何其他哑剧类型，只要它是行分隔的 JSON 格式，它的工作方式就相同。哑剧类型本身只是描述格式的约定，只要双方都理解它，就足够了。