README.md

This is a file that describes how our pipeline should be working for Marick Manrho. I hope it will be useful to him.

Resub.drawio

<mxfile host="app.diagrams.net" modified="2020-06-27T10:18:29.268Z" agent="5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.106 Safari/537.36" etag="AmuKsJVuIf-cw4v_ONBV" version="12.9.10" type="device"><diagram id="FwPP0vIwTJtMYGge0z0D" name="Page-1">3VlRk5owEP41Pl4HiIC+Kt715tppZ5z2zscoK+QmEhqiYn99gwQDMjq2YsHzRXbJhs337X4E6KHxKn3iOA6/Mh9ozzL8tIe8nmWZRt+Rf5lnl3v6hp07Ak58NUg7puQ3FJHKuyY+JJWBgjEqSFx1LlgUwUJUfJhztq0OWzJavWqMA6g5pgtM695X4osw9w4sV/s/AwnC4sqmM8zPrHAxWK0kCbHPtiUXmvTQmDMm8qNVOgaagVfgksc9njh7SIxDJC4J+PaUTn89Pz/Y76ufQYBfX17G/EEtIxG7YsHgy/Urk3ERsoBFmE60d8TZOvIhm9WQlh7zhbFYOk3pfAchdopMvBZMukKxouospES8ZeGfbGXNSme8VM28N3bKyPPMkju5fOVK2Jov4MyaizLCPABxZpx9IElWN7AVCL6TcRwoFmRTzQOrMgsO4zQT8kCR8RfEFPW/wXStLtUGVS3APrgSdhX6nRGZoWUoJUJ9BajSIWQY1SnyvFTUEXmHNK7gs0Ynh2Q9r3FaZWwbEgHTGO+B3UqFrbKzJJSOGWV8H4se5c9xDqxtgAtIz/NWx7kIOMLLcpW91SpoFtIWlhTQMU5Tc11HtNIA/6JVEmK+eysbMz1DZuqwvXUDjUMXNtuJIvhPGtfy3cf9mJSiNjlFNZ3Lss04ZjzbalkOlcmP5lweBeKAUEMKOHQ9w3UbUsCqANqt69/g/vRP98es0h43b5b+XTRL/1SzAJel293O6HesNewajklVdBoEcmS7aOjdBkjU+ibLOYVk8xXZJJDW0OkYkKZZA6ujat2g6g4uVF2nTdEd1Es8hgXBWWBM4mbVYuJ4hje5TZF34JHMuZci78CWxLyLZzLr4713uhj3YZvAm/Unpx+xjwWU9zGy1WWn69uxQaIl66xgHW9vTKd1wRrWQJ5EeE4zaEUIZWTlvkeQKEjuZs9zS3SlqT9U5K9E9eceNPkD</diagram></mxfile>

rxjs.js

const { rxObserver } = require('api/v0.3');
const { 
  timer,  
  Observable, 
  BehaviorSubject, 
  Subject, 
  ReplaySubject,
  forkJoin,
  pipe
} = require('rxjs');
const { 
  take,
  map,
  tap, 
  delay, 
  pluck, 
  filter, 
  distinctUntilChanged,
  skip
} = require('rxjs/operators');

const getFromStorage = new Observable( (observer) => {
  observer.next({
    a: "A",
    b: "B",
    c: "D"
  });
  observer.complete();
}).pipe(
  delay(1)
);

const getFromServer = new Observable( (observer) => { // Observable<Identity>
  observer.next({
    a: "A",
    c: "C",
    d: "D"
  });
  observer.complete();
}).pipe(
  delay(4)
);

const resub = new ReplaySubject(1); // Subject<Settings>
resub.pipe(
  map(val => JSON.stringify(val))
).subscribe(rxObserver("Replay"));

resub.subscribe(
  (new_settings) => {
    // localStorage.store(new_settings);
    console.log("stored in localstorage");
  }
);

const special_pipe = pipe(
  skip(1),                // Skip the local storage update
  distinctUntilChanged(), // Register the server response
  skip(1),                // Skip server response
  filter(val => !!val ),  // Skip no changes in the server setting
);

resub.pipe(
  pluck("b"),
  special_pipe,
  map(val => JSON.stringify(val)),
  tap( (val) => {
    // api.send_bio(val);
  })
).subscribe(rxObserver("Plucker bio"));

resub.pipe(
  pluck("username"),
  special_pipe,
  map(val => JSON.stringify(val)),
  tap( (val) => {
    // api.send_account(val);
  })
).subscribe(rxObserver("Plucker username"));

getFromStorage.subscribe(val => resub.next(val));

forkJoin({
  storage: getFromStorage,
  server: getFromServer
}).pipe(
  map( (val) => {
    return {...val.storage, ...val.server}
  })
).subscribe(val => resub.next(val));

timer(8).subscribe(
  () => resub.next({
    b: "New bio"
  })
);