MeruTask 同步顺序执行异步任务

demo.coffee

request = require 'request'
client= require('redis').createClient()
meru=require './libs/meruTask'

###
GET lastPostId FROM api                 (getLastPostIdFromAPI)
GET nextPostId FROM redis DEFAULTS 1    (getNextPostIdFromRedis)

LOOP:
  IF nextPostId<=lastPostId               (checkNewPostsAvailable)
    FETCH json data                       (getTalkPostAvailable)
    SAVE nextPostId TO redis
  ELSE
    WAIT 5MIN
    GET lastPostId FROM api                 (getLastPostIdFromAPI)

###

getBaseRequestOptions=()->
  proxy = 'http://127.0.0.1:8888/'
  headers=
    'Accept':'application/json, text/plain, */*'
    'Accept-Encoding':'gzip, deflate, sdch'
    'Accept-Language':'en-US,en;q=0.8'
    'Cache-Control':'no-cache'
    'Connection':'keep-alive'
    'Host':'7gogo.jp'
    'Pragma':'no-cache'
    'User-Agent':'Mozilla/5.0 (Windows NT 6.3; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/43.0.2357.2 Safari/537.36'
  gzip:true
  # proxy:proxy
  headers:headers

getTalkPostAvailable=new meru.createTask
  task:(taskStatus)->
    options=getBaseRequestOptions()
    options.qs=
      direction:'NEXT'
      limit: 100
      postId:taskStatus.nextPostId
      talkId:taskStatus.talkId
    options.url='http://7gogo.jp/api/talk/post/list'

    request options,(err,res,body)=>
      if err?
        @reject taskStatus
      else
        try
          json = JSON.parse(body)
        catch e
          console.log "FAIL parse posts list"
          @reject taskStatus
          return

        if json.nextPostId?
          taskStatus.nextPostId=json.nextPostId
        else
          taskStatus.nextPostId=json.posts[0].postId
        client.set "nextPostId:#{taskStatus.talkId}",taskStatus.nextPostId
        console.log "#{taskStatus.talkTitle} postlist nextPostId#{taskStatus.nextPostId}"
        @resolve taskStatus
  onFulfilled: (taskStatus)->@lazyTask(getNextPostIdFromRedis,taskStatus)

checkNewPostsAvailable= new meru.createTask
  task:(taskStatus)->
    if taskStatus.nextPostId<taskStatus.lastPostId
      @resolve taskStatus
    else
      @reject taskStatus

  onFulfilled:getTalkPostAvailable

  onRejected: (taskStatus)->
    @delayTask getLastPostIdFromAPI,taskStatus,10000



getNextPostIdFromRedis=new meru.createTask
  task:(taskStatus)->
    client.get "nextPostId:#{taskStatus.talkId}",(err,data)=>
      if data?
        console.log "#{taskStatus.talkTitle} nextPostId:#{data}"
        taskStatus.nextPostId=data
      else
        data =1
        client.set "nextPostId:#{taskStatus.talkId}",data

      taskStatus.nextPostId=data
      @resolve taskStatus

  onFulfilled: checkNewPostsAvailable

  onRejected: getNextPostIdFromRedis

getLastPostIdFromAPI= new meru.createTask
  task:(taskStatus)->
    options=getBaseRequestOptions()
    options.qs=
      talkIds:taskStatus.talkId
    options.url='http://7gogo.jp/api/talk/info'
    request options,(err,res,body)=>
      if err?
        @reject taskStatus
      else
        try
          json=JSON.parse(body)
        catch e
          console.log "FAIL parse talk/info/"
          @reject taskStatus
          return
        console.log "#{taskStatus.talkTitle} lastPostId:#{json.talks[0].lastPostId}"
        taskStatus.lastPostId= json.talks[0].lastPostId
        @resolve taskStatus

  onFulfilled:getNextPostIdFromRedis

  onRejected:(parentResult)->
    console.log "api catch"
    console.log taskStatus
    console.log parentResult

mainTask = new meru.createTask
  task:(taskStatus)->
    @resolve taskStatus
  onFulfilled: getLastPostIdFromAPI


client.set('nextPostId:XDXHrpvEVMS9GtN76wEuUm==','1')
client.set('nextPostId:AkENC_TJ_aS9GtN76wEuUm==','1')
meru1=new meru.createTaskAgent()
meru2=new meru.createTaskAgent()

meru1.start(mainTask,{talkTitle:'haruppi',talkId:'XDXHrpvEVMS9GtN76wEuUm=='})
meru2.start(mainTask,{talkTitle:'chiharu',talkId:'AkENC_TJ_aS9GtN76wEuUm=='})

meruTask.coffee

class MeruTaskAgent
  start:(entryTask,taskStatus)->
    entryTask.do(taskStatus)


class MeruTask
  constructor: (taskDef) ->
    {@task,@onFulfilled,@onRejected}=taskDef

  resolve:(taskStatus)->
    resolveType=@onFulfilled.constructor.name
    switch resolveType
      when 'MeruTask'
        @onFulfilled.task(taskStatus)
      when 'Function'
        @onFulfilled(taskStatus)

  reject:(taskStatus)->
    rejectType=@onRejected.constructor.name
    switch rejectType
      when 'MeruTask'
        @onRejected.task(taskStatus)
      when 'Function'
        @onRejected(taskStatus)

  delayTask:(task,taskStatus,timeout)->
    console.log "not found new post wait for #{timeout}ms"
    setTimeout ()=>
      task.task(taskStatus)
    ,timeout

  lazyTask:(task,taskStatus)->
    do ()->
      task.task(taskStatus)

  do:(taskStatus)->
    @task.call(@,taskStatus)

exports.createTask =(params)->
  new MeruTask(params)

exports.createTaskAgent=(entryTask,taskStatus)->
  new MeruTaskAgent(entryTask,taskStatus)