Sagas Yielding to each other asycnronously

[redux-saga]

Sagas Yielding to each other asycnronously

[bradennapier]

Cool, this is opening a lot of options with sagas! Here is another example where we can make a promise resolve with an error as well as utilizing the HTML5 GeoLocation API and a little bit different styling of promise handling

function userPositionPromised() {
  const position = {}
  if (navigator.geolocation) {
    navigator.geolocation.getCurrentPosition (
      location  => position.on({location}),
      error     => position.on({error}),
      { enableHighAccuracy: true }
    )
  }
  return { getLocation: () => new Promise(location => position.on = location) }
}

function* getUserLocation() {
  const { getLocation } = yield call(userPositionPromised)
  const { error, location } = yield call(getLocation)
  if (error) {
    console.log('Failed to get user position!', error)
  } else {
    console.log('Received User Location', location)
  }
}

export default function* root(getState) {
  // Fork each required daemon on startup, for now only one
  yield fork(getUserLocation)
}

[yelouafi]

@bradynapier thanks for sharing this.

You may find a related SO question here

Basically, whenever we want to interface with a push source (DOM events, websocket messages); we build an event iterator (somewhat like channels in CSP or coroutines) then we can iterate over it inside the Saga

here is a slightly version of your example

• We buffer incoming actions when the Saga is not waiting for them
• We define a [CANCEL] method on the returned promises, so redux-saga will automatically call this
  method if the promise looses in a race

Note: Didn't test the code

import { CANCEL } from 'redux-saga'

// define a cancellable promise
// will be called by redux-saga to cancel this promise
// upon loosing on race
function cancellablePromise(p, doCancel) {
  p[CANCEL] = doCancel
  return p
}

function firebaseIterator(ref) {
  const messageQueue = []
  const resolveQueue = []

  const listenerID = ref.on('value', snapshot => {
    const msg = snapshot.val()
    // anyone waiting for a message ?
    if (resolveQueue.length) {
      const nextResolve = resolveQueue.shift()
      nextResolve(msg)
    } else {
      // no one is waiting ? queue the event
      messageQueue.push(msg)
    }
  })

  function close() {
    ref.off(listenerID)
  }

  return {
    getNext() {
      // do we have queued messages ?
      if(messageQueue.length) {
        return cancellablePromise( 
           Promise.resolve(messageQueue.shift()) 
        )
      } else {
         return cancellablePromise(
            new Promise(resolve => resolveQueue.push(resolve))
        )
     }
    }
  }
}

function* onFirebaseData(accountRef, path, getState) {
  const localRef = accountRef.child(path)
  const { getNext } = yield call(firebaseIterator, localRef)

  var cancel, data
  while (!cancel) {
    // if getNext loose, redux-saga will try to cancel it 
    // by looking for a CANCEL method on its promise result
    ({data, cancel} = yield race({
      data: call(getNext),
      cancel: take(action => action.type === CANCEL_LISTENER && action.path === path)
    })

    if (data) {
      console.log('Received Data from Firebase Path!', path)
      console.log(data)
    }
  }
  console.log('Cancelling Firebase Listener at Path: ', path)
}

[bradennapier]

Ahh, shortly after this I did find a post about it and was able to make this work. I think this would belong in the documentation somewhere as it's likely a common need. Perhaps even some sort of pattern officially supported may be nice?

Anyway, not sure I did it in the best manner possible, but here is the working results for anyone whom it may help:

How It Works: It listens to global paths once an account has been authenticated. It doesn't do anything with the response yet but the idea is to listen to data that is needed anywhere the user is on the app and handle the logic based on the data received from Firebase. It returns the getNext and the listenerID (which is used to cancel the callback if needed for some reason).

Race is used to allow for cancelling by listening for a cancellation action and seeing if the paths match. Probably will need to enhance that a bit.

Awesome Stuff!

function registerListenerPromised(ref) {
  var listener
  const listenerID = ref.on('value', snapshot => {
    if (listener) {
      listener.on(snapshot.val())
      listener = null
    }
  })

  return {
    listenerID,
    getData() {
      if(!listener) {
        listener = {}
        listener.promise =
          new Promise(response => listener.on = response)
      }
      return listener.promise
    }
  }
}

function* onFirebaseData(accountRef, path, getState) {
  const localRef = accountRef.child(path)
  const { getData, listenerID } = yield call(registerListenerPromised, localRef)
  // set this to false at any time to cancel this listener
  var active = true
  while (active) {
    const {data, cancel} = yield race({
      data: call(getData),
      cancel: take(action => action.type === CANCEL_LISTENER && action.path === path)
    })
    if (cancel) { var active = false } else {
      console.log('Received Data from Firebase Path!', path)
      console.log(data)
    }
  }
  console.log('Cancelling Firebase Listener at Path: ', path)
  localRef.off(listenerID)
}

function* setupGlobalListeners(getFirebaseRef, getState) {
  const accountRef = getFirebaseRef()

  // We can't use options like forEach and map in generators
  for (let i in GLOBAL_PATHS) {
    const path = GLOBAL_PATHS[i]
    yield fork( onFirebaseData, accountRef, path, getState )
  }
}

function* waitForAuth(getFirebaseRef, getState) {
  // We must wait for the firebase account to be authenticated elsewhere
  yield take(ACCOUNT_READY)
  yield fork(setupGlobalListeners, getFirebaseRef, getState)
}

export default function* root(getState) {
  const getFirebaseRef = () => getState().auth.firebase
  yield fork(waitForAuth, getFirebaseRef, getState)
}

Feel free to let me know if I made any mistakes, still learning this pattern but reminds me of good ol' coroutines from tcl days.

[bradennapier]

Thank you! So I spent the entire day playing with this just to see what might come out of it. I really like this pattern. In my mind one thing that is needed is a better way to handle asynchronous events as discussed. This provides a way to do it but can be cumbersome if doing often, so I wanted to try to create a pattern which could help to automate this a bit... I am no pro at javascript but I do generally have a grip on coroutine-style programming.

I am sure I am doing a lot of things completely wrong here, but the end result seems to be a pattern which is re-useable and very easy to work with. It's a lot to read, but it seems to be a good start - would like an honest opinion if possible and would be happy to share more of it if desired:

function* waitForAuth(getState) {
  while (true) {
    yield take(ACCOUNT_READY)
    yield fork(setupGlobalListeners, getState)
  }
}

export default function* root(getState) {
  yield fork(waitForAuth, getState)
}

Simple enough, I await the action telling me the authentication is available...

function* setupGlobalListeners(getState) {
  const accountRef = getState().auth.firebase
  const globalKeys = Object.keys(GLOBAL_LISTENERS)

  for (let key of globalKeys) {
    const { path, types } = GLOBAL_LISTENERS[key]
    const Saga = new SagaState(path, getState)
    yield fork(generateObserver, Saga)
  }
}

I am using classes to generate a react-like pattern for handling redux actions and only dispatching properly handled data to the store...

function* generateObserver(Saga) {
  console.log('Process Saga!', Saga)
  var cancel, saga
  while(!cancel) {
    ({saga, cancel} = yield race({
      saga:     apply(Saga, Saga.sagaShouldMount),
      cancel:  apply(Saga, Saga.__checkStatics__)
    }))
  }
  console.log('Done')
}

I am attempting to provide cancel however at this time the cancel does not appear to work as desired.

Here's the look at what an instance of SagaClass looks like:

@onStore('firebase')
class Example extends SagaState {

  static cancelTypes = [
    'CANCEL_SAGACLASS'
  ];

  static monitorTypes = [
    'FIREBASE_UPDATED'
  ];

  shouldSagaCancel(action) {
    console.log('Should I Cancel?')
    console.log(action)
    return true
  }

  onMonitoredType(action) {
    console.log('Monitor Executed!  This shows listening for an action and responding, in this case when the firebase is updated by the observer')
    console.log('Action: ', action)
  }

  async observer() {
    const { getNext, on } = this.createObservable('firebase')
    const ref = this.getStore().ref.child('unadopted')
    const observerID = ref.on('value', on)

    while (!this.cancel) {
      const data = await getNext()
      this.dispatch({type: 'FIREBASE_UPDATED', data: data.val() })
    }
    ref.off('value', observerID)
  }

  * sagaRender() {
    this.observer()
    yield take(action => action.type === 'LOGOUT_CLEAR_TOKEN')
    this.dispatch({type: 'GOODBYE!'})
  }

}

export default Example

So there's more to this but I thought I would just post the example... essentially this does a few things... Keeping in mind I took all the extra logic out of handling the data in this case since it would muddy up the example with firebase-specific logic...

1. You define in the decorator what part of the state this class will be responsible for. You receive this.getStore() which allows you to get that part of the store at any point in time. My goal would be to somehow restrict this in a way there are less side effects.
2. You define two statics, if desired... cancelTypes and monitorTypes - these allow you to define actions that you want to receive whenever they occur. In both cases the action is given as an argument. You can define multiple types in the array.

cancelTypes: With cancelTypes the "shouldSagaCancel" is called and you return true or false with true resulting in the termination of the observer. In the example, the SagaClass will be cancelled upon the first time any of the cancel types are received.

monitorTypes: With monitorTypes you receive these in "onMonitoredType" and can handle this in any way you desire.

1. We move outside of redux-saga in some parts but redux-saga is still controlling everything. For example, we can have redux-saga do a put(ACTION) simply by calling this.dispatch(ACTION) whenever we need

Now to the "magic" - i wanted to make it super easy to handle push callbacks in this type of environment, so I came up with the following pattern... I am not sure if this ends up giving any kind of issue, but here goes:

Using an async function rather than a generator (just was simpler at the end of the day, however I still have to handle the cancellation properly which I haven't done yet), we can call to capture an "observer" object which includes the following elements:

const { getNext, on } = this.createObservable('firebase')

(note that the 'firebase' in the create observable is just because we generate a reference using Symbol() at the moment so it's mainly for debugging that)

we then call the push-style callback and provide the observers "on" value to it as the callback. In this example, firebase returns an ID which can be used to cancel the listener:

const observerID = ref.on('value', on)

Now we simply use the await command to wait for a resolution and loop it so we receive future iterations (firebase returns a ref and you use data.val() to capture it's actual value) :

while (!this.cancel) {
     const data = await getNext()
     this.dispatch({type: 'FIREBASE_UPDATED', data: data.val() })
}

What's beautiful is how simple the pattern is to reason about the cancellation (once I get that working fully)

async observer() {
    const { getNext, on } = this.createObservable('firebase')
    const ref = this.getStore().ref.child('unadopted')
    const observerID = ref.on('value', on)
    while (!this.cancel) {
      const data = await getNext()
      this.dispatch({type: 'FIREBASE_UPDATED', data: data.val() })
    }
    ref.off('value', observerID)
  }

there's a lot going on to make this work that isn't shown here, but here is the createObservable from the parent class... i am sure it will look familiar to you :-)

createObservable(name) {
    const actionQueue = []
    const dispatchQueue = []
    const observerRef = Symbol(name)

    this[observerRef] = response => {
      if (dispatchQueue.length) {
        const nextDispatch = dispatchQueue.shift()
        nextDispatch(response)
      } else {
        actionQueue.push(response)
        console.log(actionQueue)
      }
    }
    return {
      on: this[observerRef],
      getNext() {
        if(actionQueue.length) {
          return cancellablePromise(
            Promise.resolve(actionQueue.shift())
          )
        } else {
          return cancellablePromise(
            new Promise(resolve => dispatchQueue.push(resolve))
          )
        }
      }
    }
  }

What the final result looks like as I modify data in firebase (note in the code when i ran this i had NOTIFICATIONS_READY as the cancelType)... I also changed it so i returned false when the shouldSagaCancel was triggered: