amitrahav

import React from 'react'
import { connect } from 'react-redux';
import MessagesActions from './store/messages/MessagesActions';
class SSEComponent extends React.Component{
    componentDidMount(){
        // Open SSE connection with server - listen messages for the first index video
        this.props.openVideoConnection(1)
    }
    render(){
        return (

amitrahav

import MessagesActionTypes from './MessagesActionTypes';
import SSEConstants from '../sse/sseConstants';
const videoMessages = (indexIdentifier) => ({
  type: SSEConstants.SSE_REQUEST,
  payload: {indexIdentifier},
  url: `videos/subscribe`,
  baseAction: MessagesActionTypes.MessagesForVideo,
  stopSreamingOn: {
    key: "progress",
    value: "100"

amitrahav

import MessagesActionTypes from "./MessagesActionTypes";
const initialState = {
  allVideos: [],
};
const MessagesReducer = (state, action) => {
  if (!state) {
    return initialState;
  }
	const {type, payload} = action;
  switch(type){

amitrahav

import sseMiddleware from "./sse/SSEMiddleware";
const store = createStore(persistedReducer, applyMiddleware(sseMiddleware, ...otherMiddelwares));
const redux = { store, ...otherStuffIfNeeded };
export default redux;

amitrahav

import SSEConstants from "./SSEConstants";
const sseMiddleware = ({ dispatch }) => {
	// This middleware will handle all requests, so we need to identify only the sse ones
  return (next) => async (action) => {
		// The way of desiding if it's a sse request we need to handle - is by the action type
    if (action.type !== SSEConstants.SSE_REQUEST) {
      return next(action);
    }
    const { url, baseAction, payload, stopSreamingOn, chanel } = action;
  	// All sse type action should indlude all of those keys:

amitrahav

const BASE_URL = "<https://localhost:8000/api/v1/>"; // My server base url 
const SSE_REQUEST = "SSE_REQUEST"; // This will be the identifier for the middlware to process that request.
// All states of requests the reducers will need to know
const createSSERequestAction = (actionType) => ({
  OPEN: `${actionType}_OPEN`,
  CLOSE: `${actionType}_CLOSE`,
  RECIVED: `${actionType}_RECIVED`,
  FAILURE: `${actionType}_FAILURE`,
  CONNECTING: `${actionType}_CONNECTING`,
  BASE: actionType,

amitrahav

from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse
from typing import AsyncGenerator
import asyncio
import json 
# Creating my app
app = FastAPI()
progress_bar = ["10", "30", "50", "60", "80", "100"]
async def subscribe() -> AsyncGenerator:
    for progress in progress_bar: # This function in real can be a redis subscribtion

amitrahav

// Listen for possible errors
evtSource.onerror = function() {
  console.log("EventSource failed.");
};
// Listen for possible errors
socket.addEventListener("error", function (event) {
  console.log("WebSocket error:", event);
});

amitrahav

eventSource.close();
socket.close();

amitrahav

evtSource.onmessage = (e) => {
  console.log(e.data)
};