How to create subset of observable basing on input argument?

user_2446799 0
Viewed 101

I know the title might sound confusing, but I'm not sure how to express it accurately.

Let me explain the use case:

We've got a calendar with recurring events (potentially infinite list of events repeating eg every Monday forever)

I've got a bunch of UI elements interested in some part of those events (eg. some needs events from next week, some from next month, some from the previous year - they all might be displayed at once). Note that those requirements might overlap - the eg. current week and current month actually include an overlapping set of days, but those UI components are not aware of each other.

I need to refresh those events on some actions in the app (eg. the user changed calendar permission, list of calendars in his system, user minimized and re-opened the app) - I've got streams of those events like $appStatus $permissionStatus, $selectedCalendars

Fetching events is expensive and I want to re-use fetching and existing data as much as possible. Eg. if 3 components need events from the same week, I don't want to make 3 requests.

Also if one component requests events from March and another from 2nd week of March, I'd like to solve it with one request if possible. Note they might start their request at different moment.

I also don't want to fetch events that are not requested anymore (eg. part of UI showing events from the previous month is not rendered anymore)

I've got a few ideas about how to do it in plain JS, but I wonder if it's possible to create RxJs-ish solution for that

My raw-js-rx-mix ideas:

When a new range is requested, I 'round' it to full weeks. Then I create new observable with shared value for each week and remember it in some map

eg

const observablesMap = new Map();

function getObservableForDate(date: Date) {
  const weekStart = getStartOfWeek(date);
  // if we have 'cached' version of observable for this week - return cached one
  if (observablesMap.has(weekStart.getTime()) {
    return observablesMap.get(weekStart.getTime())
  }

  const weekObservable = // create observable that shares results

  // set it to the map
  // return it
}

But after a few hours of research, I have no idea if and how would I implement it in RxJS way.

Let's assume the fetching function signature is fetchEvents(startDate, endDate)

ps. I don't expect a working solution code, but just some guide. I've checked most of RxJS documentation and could not find anything promising for such use case

rxjs reactive-programming javascript
1 Answers

I created a running rxjs solution on stackblitz. I will also add the code here in case stackblitz shuts down one time in the future.

General idea

  • Save all requests and decide if you need a new request or not
  • Depending on previous decicion fake or process http request
  • Depending on the previous kind of request find the existing request or return the newly requested one.

If there need to be further infos please let me know and I try to explain in detail or add comments. I did not implement 100% of your requirements but with the following solution as base it should be possible to expand interfaces and functions within the pipes to implement them. Also if you see any code redundancy or optimization to interfaces, let me know and I will adapt

Interfaces

interface Requests {
  action: Action,
  currentRequest: number,
  accumulatedRequests: number[],
}

interface FulfilledRequest extends Requests{
  httpRequest: string
}

interface Response {
  accumulatedHttpResponses: {
    request: number,
    response: string
  }[],
  response: string
}

Default Values

const defaultRequests: Requests = {
  action: Action.IgnoreRequest,
  currentRequest: -1,
  accumulatedRequests: []
}

const defaultResponseStorage: Response = {
  accumulatedHttpResponses: [],
  response: ''
}

Enum

enum Action {
  IgnoreRequest,
  ProcessRequest
}

Functions

const isUpdateAction = (action: Action) => action === Action.ProcessRequest

const fakeHttp = (date: number): Observable<string> => of('http response for: ' + date).pipe(
  tap(v => console.warn('fakeHttp called with: ', v))
);

const getResponseForExistingRequest = (storage: Response, request: FulfilledRequest): Response => {
  const index = storage.accumulatedHttpResponses.findIndex(response => response.request === request.currentRequest);
  return {
    accumulatedHttpResponses: storage.accumulatedHttpResponses,
    response: storage.accumulatedHttpResponses[index].response
  }
}

const getResponseForNewRequest = (storage: Response, request: FulfilledRequest): Response => {
  const newEntry = {request: request.currentRequest, response: request.httpRequest};
  return {
    accumulatedHttpResponses: [...storage.accumulatedHttpResponses, newEntry],
    response: request.httpRequest
  }
}

const getIgnoredRequest = (date: number, requests: Requests): Requests => ({
  currentRequest: date,
  action: Action.IgnoreRequest,
  accumulatedRequests: requests.accumulatedRequests
})

const getProcessedRequests = (date: number, requests: Requests): Requests => ({
  currentRequest: date,
  action: Action.ProcessRequest, 
  accumulatedRequests: [...requests.accumulatedRequests, date]
})

const processRequest = (requests: Requests, date: number): Requests => {
  const requestExists = requests.accumulatedRequests.some(request => request === date);
  return requestExists
    ? getIgnoredRequest(date, requests)
    : getProcessedRequests(date, requests)
  }

const processFulfilledRequest = (storage: Response, request: FulfilledRequest): Response => isUpdateAction(request.action)
  ? getResponseForNewRequest(storage, request)
  : getResponseForExistingRequest(storage, request)


const fulfillFakeRequest = (requests: Requests): Observable<FulfilledRequest> => of('').pipe(
  map(response => ({...requests, httpRequest: response})),
)

const fulfillHttpRequest = (requests: Requests): Observable<FulfilledRequest> => fakeHttp(requests.currentRequest).pipe(
  map(response => ({...requests, httpRequest: response}))
)

Final Connection via Observables

const date$: Subject<number> = new Subject();

const response$ = date$.pipe(
  scan(processRequest, defaultRequests),
  switchMap((requests): Observable<FulfilledRequest> => isUpdateAction(requests.action)
    ? fulfillHttpRequest(requests)
    : fulfillFakeRequest(requests)
  ),
  scan(processFulfilledRequest, defaultResponseStorage),
  map(response => response.response)
)
user_63247750
Related