SSE Spring and Angular

content.app.component.ts

import { Routes } from '@angular/router';
import { BehaviorSubject, Observable } from 'rxjs/Rx';
import { Component, Injectable, NgZone, OnInit } from '@angular/core';

const EventSource: any = window['EventSource'];

@Injectable()
export class Sse {

  constructor(private zone: NgZone) { }

  get(sseUrl: string): Observable<any> {
    return new Observable<any>(obs => {
      const es = new EventSource(sseUrl);
      es.onmessage = evt => {
        const data = JSON.parse(evt.data);
        this.zone.run(() => obs.next(data));
      };
      return () => es.close();
    });
  }

}

export interface Content {
  id: number;
  type: string;
  content: string;
}

@Injectable()
export class ContentService {

  private content$ = new BehaviorSubject(null);

  constructor(private sse: Sse) {
    this.sse.get('/api/v1/content/stream').subscribe(data => {
      this.content$.next(data);
    });
  }

  findAll() {
    return this.content$.share();
  }

}

@Component({
  selector: 'app-content-table',
  template: `
<table>
  <tr>
    <td>#ID</td>
    <td>Type</td>
    <td>Content</td>
  </tr>
  <tr *ngFor="let content of contents">
    <td>{{content.id}}</td>
    <td>{{content.type}}</td>
    <td>{{content.content}}</td>
  </tr>
</table>`
})
export class ContentTableComponent implements OnInit {

  contents: Content[];

  constructor(private contentService: ContentService) { }

  ngOnInit() {
    this.contentService.findAll().subscribe(data => {
      this.contents = data;
    });
  }

}

@Component({
  template: 'Empty Page...'
})
export class StartPageComponent {

}

@Component({
  template: `
    <app-content-table></app-content-table>
    <app-content-table></app-content-table>
  `
})
export class ContentPageComponent {

}

export const appRoutes: Routes = [
  {path: '', component: StartPageComponent},
  {path: 'content', component: ContentPageComponent},
];

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent {

}

ContentController.java

@RequestMapping("/api/v1/content")
@RestController
public class ContentController {

	@Autowired
	ContentRepository contentRepository;
	
	private final List<SseEmitter> emitters = new ArrayList<>();
	
	@RequestMapping(path = "/stream", method = RequestMethod.GET)
    public SseEmitter stream() throws IOException {
        SseEmitter emitter = new SseEmitter();
        emitter.send(this.contentRepository.findAll(), MediaType.APPLICATION_JSON);
        emitters.add(emitter);
        emitter.onCompletion(() -> emitters.remove(emitter));
        return emitter;
    }
	
	@RequestMapping(method = RequestMethod.POST)
	public void create(@RequestBody Content content) {
		this.contentRepository.save(content);
		emitters.forEach((SseEmitter emitter) -> {
            try {
                emitter.send(this.contentRepository.findAll(), MediaType.APPLICATION_JSON);
            } catch (IOException e) {
                emitter.complete();
                emitters.remove(emitter);
                e.printStackTrace();
            }
        });
	}
	
	@RequestMapping(method = RequestMethod.GET)
	public Iterable<Content> get() {
		return this.contentRepository.findAll();
	}
	
}