Suggestion on Handling errors in Channels about channel HOT 4 CLOSED

Great question! It inspired me to make a change.

As of 0.6.3 you can send promises through channels:

const channel = Channel();

(async () => {
  try {
    const devcorn = await channel.shift();
  } catch (exception) {
    console.error(`Sorry, no unicorns today.`);
  }
})();

await channel.push(
  fetch(`https://twitter.com/chjango/status/925787871973400576`)
);

Now you can find out idiomatically whether there's an error with each datum. Does that solve your problem?

from channel.

Hmm, I'm not sure. What I really want to have is a single promise on the larger operation populating the channel, not each channel item.

I've solved this myself for now like this: (sorry, I'm not very es7 fluent yet)

search_channel = (query) -> Promise.try () ->
  ch = Channel(100)
  search_op = new Promise (resolve, reject) ->
     handle_response = (err, response) ->
        if err?
           ch.close()
           return reject( err )
       
        for hit in response.hits
            ch.push hit

        if response.more?
             some_client.search_next( 100, handle_response)
        else
             ch.close()
             return resolve()
     some_client.search( query, 100, handle_response)
      
  return {
    channel: ch.readOnly()
    result: search_op
  } 

Then in my main code:

search_channel( "foo=bar")
.then (s) ->
   s.channel.forEach (item) ->
       # process each item
  .then () ->
      s.result.then () ->
         # Fetch completed successfully
      .catch (err) ->
         # Fetch had an error

Essentially the channel always gets closed, and after the forEach I check the result promise to ensure whether or not I got everything.

from channel.

Just an update here, I do like the ability to send a rejection via the channel as it allows my consumer to process all items that do get pushed to the channel up until it hits an error. This simplifies my code a lot. Here's a real Elasticsearch fetcher function that returns a channel

exports.search_channel = (query, ch=Channel(100)) ->
  do () ->
    query.scroll = '30s'
    query.size = ch.length
    total = 0
    loop
      try
        response = await if total == 0
          log.debug "initial fetch"
          es_client.search query
        else
          log.debug "fetching more"
          es_client.scroll {
            scrollId: response._scroll_id
            scroll: '30s'
          }

        for hit in response.hits.hits
          await ch.push hit._source
          total += 1
      catch err
        await ch.push Promise.reject err
        break

      break unless response.hits.total > total
    ch.close()

  ch.readOnly()

The consumer for this can work like this:

channel = es.search_channel { index: "foo", query: "*"}
try
   await channel.forEach (record) ->
     # process record
catch err
   # Fetch error

I guess the only thing this lacks is the ability to throw a channel exception that would allow the consumer to stop processing as soon as there is an error, regardless of how many items are in the channel.

from channel.

.forEach should stop and return a rejected promise immediately if there are any errors in the channel so your consumer code above should work as written.

from channel.