I use Mercure with Symfony4 and Angular9. On the click event of a button, I publish the Mercure update :

In the component.html :

<button (click)="insertSynonyme()" class="btn btn-default mb-2">Save</button>

In the component.ts :

constructor(
    ...
    private mercureServ: MercureService,
    ...
  ) {
    ...
  }
...
insertSynonyme(){
    const data = {
      old_cle: this.synoForm.value.cle_syno,
      add_synonymes: this.dataSynonyme.oldSynonyme
    };
    // make database update first
    this.generalServ.insertSynonyme(data).subscribe( // calls a Symfony API
      res => {
               this.mercureServ.getServerSentEvent('http://localhost:3000/.well-known/mercure','suggestioncontenu').subscribe(
                   data => {
                        console.log("==================== dans callback de getServerSentEvent depuis component ts, on a data = ", data);
                        this.mercureServ.close();
                   }
               );
             });
 }

The insertSynonyme code :

insertSynonyme(data){
    return this.dataServ.postData(API.insertSynonyme, data);
  }

The postData code :

@Injectable({
  providedIn: 'root'
})
export class DataService {
  constructor(protected http: HttpClient) { }

  postData(url, dataObject): Observable<any> {
    const httpOptions = {
      headers: new HttpHeaders({
        'Accept': 'application/json',
        'Content-Type': 'application/json',
        'Authorization': 'Bearer ' + sessionStorage.getItem('token')
      })
    };
    return this.http.post(ws_URL.concat(url), dataObject, httpOptions).pipe(retry(1));
  }
  ...
}

The Symfony API called :

use AppUtilsConstantSrv;
use FOSElasticaBundleElasticaClient;
use FOSElasticaBundleFinderFinderInterface;
use FOSElasticaBundleFinderTransformedFinder;
use FOSRestBundleControllerAnnotations as Rest;
use FOSRestBundleControllerAnnotationsQueryParam;
use FOSRestBundleRequestParamFetcher;
use SymfonyBundleFrameworkBundleControllerAbstractController;
use SymfonyComponentHttpFoundationJsonResponse;
use AppServiceSearchService;
use SymfonyComponentHttpFoundationRequest;
use SymfonyComponentHttpFoundationSessionSessionInterface;
use SymfonyContractsTranslationTranslatorInterface;
use SymfonyComponentHttpFoundationResponse;
use SymfonyComponentRoutingAnnotationRoute;
use SymfonyComponentMercurePublisherInterface;
use SymfonyComponentMercureUpdate;

class SearchController extends AbstractController
{
    ...
    /**
     * @RestPost("/api/insertUpdateSynonyms", name ="api_insert_update_synonyms")
     * @RestRequestParam(name="old_cle", nullable=true)
     * @RestRequestParam(name="add_synonymes", nullable=true)
     * @RestRequestParam(name="new_cle", nullable=true)
     * @RestRequestParam(name="new_synonymes", nullable=true)
     *
     * @return Response
     */
    public function insertUpdateSynonyms(ParamFetcher $paramFetcher, SearchService $searchService, PublisherInterface $publisher)
    {
        try {
            $searchService->insertUpdateSynonyms($paramFetcher);
            $update = new Update(
                'suggestioncontenu',
                json_encode(['etat' => 'OutOfStock'])
            );
            $subscriber_id = $publisher($update);
            return new Response('');
        } catch (Exception $e) {
            return new Response('');
        }
    }

}

The MercureService in mercure.service.ts :

import { Injectable, NgZone } from "@angular/core";
import { Observable } from "rxjs";

@Injectable({
    providedIn: 'root'
})
export class MercureService {

    eventSource: EventSource;

    constructor(private zone: NgZone) { }

    private getEventSource(url: string, topic: string): EventSource {
        const urls = new URL(url);
        urls.searchParams.append('topic', topic);
        return new EventSource(urls.toString());
    }

    /**
     * return the event source stream
     */
    getServerSentEvent(url: string, topic: string): Observable<MessageEvent> {

        return new Observable(observer => {

            this.eventSource = this.getEventSource(url, topic);

            this.eventSource.onopen = op => {
                console.log("--------------------- connex entre mercure et angular est ouverte");
            }

            this.eventSource.onmessage = event => {
                const data = event.data;
                const toSend = data.substring(data.indexOf('GMT') + 3, data.length - 1).trim();
                console.log("================= event data depuis angular service = ", toSend);
                this.zone.run(() => observer.next(toSend));
            };

            this.eventSource.onerror = err => {
                console.log("================ erreur : ", err);
            }

        });
    }

    close() {
        if (this.eventSource) {
            this.eventSource.close();
        }
    }

}

At runtime, when I click the button for the first time then no data is sent, I do not see any console log showing ================= event data depuis angular service; but when I click the button again then data is sent, and console log is displaying with the data; and even if I reclick the button many times then data is sent. So why are there no data sent at the first click ?

Source: Symfony4 Questions