How to work with non-fiber asynchronous flows?

[jpsear]

Hey,

Just bringing up this issue again, prior reference here: #16

Behaviour with asynchronous code doesn't seem to work for me. Take this simple example:

import fetch from 'node-fetch'

const job = () => {
  console.log('Running job in queue')
  return new Promise((resolve, reject) => {
    fetch('https://reqres.in/api/users?page=2')
      .then(res => res.json())
      .then(json => {
        setTimeout(() => {
          resolve(json)
        }, 4000)
      })
      .catch(error => {
        reject(error)
      })
  })
}

const queue = function () {
  console.log('Running queue')
  const _this = this
  job()
    .then(result => {
      console.info('got a result', result)
      _this.success(result)
      return
    })
    .catch(error => {
      console.warn('got a error', error)
      _this.failure(error)
      return
    })
}

Jobs.register({ queue })
Jobs.run('queue')

When I start my Meteor server, I then see:

I20180218-18:18:39.886(0)? Running queue
I20180218-18:18:39.909(0)? Running job in queue
I20180218-18:18:39.990(0)? Running queue
I20180218-18:18:39.991(0)? Running job in queue
I20180218-18:18:40.081(0)? Running queue
I20180218-18:18:40.082(0)? Running job in queue
I20180218-18:18:40.170(0)? Running queue
I20180218-18:18:40.171(0)? Running job in queue
I20180218-18:18:40.262(0)? Running queue
I20180218-18:18:40.263(0)? Running job in queue

... and so on... Until I stop the server. So it's stuck in a loop.

Pretty sure the this context is not an issue here — but I could be wrong!

[msavin]

Can you post job documents history? JobsInternal.Utilities.Collection.find().fetch()

Also, what version are you on?

In the latest version - if your job is not resolved with this.success or this.failure, the queue will stop and post a message to console.

[jpsear]

Version: msavin:sjobs@2.0.0

I'm not quite sure what I'm logging, JobsInternal.Utilities.Collection.find().fetch() returns can't read property find of undefined

[msavin]

My apologies - typo: JobsInternal.Utilities.collection.find().fetch()

Can you try 2.0.1 and see what happens?

Are you getting got a error or got a result anywhere?

[ramijarrar]

If you want to use Promises you'll need to wrap the job in a fiber-aware await.

For example, this will resolve correctly as a success/failure:

import { Jobs } from 'meteor/msavin:sjobs';
import { Promise } from 'meteor/promise';
import fetch from 'node-fetch'

function registerJob({ name, run }) {
  Jobs.register({
    [name](...parameters) {
      /* Forward context/params to 'run' method - execute in a fiber-aware await to
       allow regular async syntax in provided function */
      Promise.await(run.apply(this, parameters));
    },
  });
}

registerJob({
  name: 'queue',
  async run() {
    try {
      console.log('Running queue');
      const result = await fetch('https://reqres.in/api/users?page=2');
      const json = await result.json();
      console.info('got a result', json);
      this.success(result);
    } catch (error) {
      console.warn('got a error', error);
      this.failure(error);
    }
  },
});

Jobs.run('queue');

[msavin]

@ramijarrar thanks for helping out

Btw - every job is automatically wrapped in try .. catch and it automatically runs this.failure() in the catch statement.

[jpsear]

Interesting stuff @ramijarrar. I've actually shifted to a different library (for other reasons), but hopefully this thread is helpful to others :)

[msavin]

@jpsear can you share the reasons? I'm curious :)

[jpsear]

Yeah, of course. I went with Agenda.

• Docs are great
• Concurrency control, per queue
• Example of how to set it up in an actual project
• Mongo
• Priorities
• Human language job running (every('1 day', ...))

[derouck]

Thanks for posting @ramijarrar, your sample helped me out after I had refactored some things with async await!

[msavin]

@ramijarrar would this be easy to implement into package core?

Here's what executes the jobs:
https://github.com/msavin/SteveJobs..meteor.jobs.scheduler.queue.background.tasks/tree/master/package/server/imports/actions/execute

[gary-menzel]

I have implemented this (or something similar) but have not really looked at my code for a few weeks. I think I have "specific" things in my code that would need to be made more generic. However, I don't see any reason why it could not be implemented. Maybe as a separate function (like registerJobWithPromise).

[msavin]

@gary-menzel might it be possible to automatically detect if a function a async, and treat it as such?

https://stackoverflow.com/questions/38508420/how-to-know-if-a-function-is-async

[gary-menzel]

I haven't really tried to detect them. Generally, if a function supports both a "callback" style or a "promise" the presence/absence of the callback parameter is used to trigger either logic.

I also tend to like the "explicit" way of doing things - so the name I suggested (or something similar) appeals to me from that perspective. There is also the Meteor.bindEnvironment to think about as well. That could potentially be another method (registerJobWithMeteorEnvironment).

The other possibility is that they are configuration flags in the options parameter.

While Promise is here to stay, the async and await mechanism is still being bedded down in the ECMAScript specification (albeit being stable for production use, the underlying implementation could change over time).

From my quick look at the article all the methods mentioned to detect the function as being async work in Google Chrome (which I think is still on the V8 engine - as it node.js) - so it might work.

[msavin]

@gary-menzel there must be a way to make the package smart enough to detect how to run the function

[gary-menzel]

I've got a backlog of work to do right at the the moment but I'll add "detecting a Promise" on the list there and see if I can come up with something. Currently, I am running my own copy of sjobs and will be able to experiment (as I am already running with Promises).

It might be a few weeks before I can get to it though.

[msavin]

@gary-menzel how's your availability now?

Something like this hit me this morning...

Jobs.register({
	"sendEmail": {
		type: "async",
		function: function () {
			// ...
		},
		checkEvery: 10000 // checks every 10 seconds
		moreSettings: "yay" 
	}
})

Jobs.register() would check if sendEmail is a function or object. If its a function, it would behave as it does now, and if its an object, it would run through the custom configuration.

[msavin]

Closing this thread out. @ramijarrar thanks again for your tip, I added it to the new documentation

[gary-menzel]

Hi @msavin (Max) - sorry, I have just been swamped - lost our Junior Dev on my "day job" so didn't get a chance to respond. I think your suggested solution is the simplest way. I will need to look at how it fits in with what I have done already.

I do think it would be possible to detect it - but I also think the explicit approach is equally valid.