import gevent
import msgpack
import redis
import time

from lymph.core.interfaces import Interface
from lymph.core.decorators import rpc


class Scheduler(Interface):
    service_type = 'scheduler'
    schedule_key = 'schedule'

    def __init__(self, *args, **kwargs):
        super(Scheduler, self).__init__(*args, **kwargs)
        self.redis = redis.StrictRedis()

    def on_start(self):
        self.container.spawn(self.loop)

    @rpc()
    def schedule(self, channel, eta, event_type, payload):
        self.redis.zadd(self.schedule_key, eta, msgpack.dumps({
            'event_type': event_type,
            'payload': payload,
        }))
        channel.ack()

    def loop(self):
        while True:
            pipe = self.redis.pipeline()
            now = int(time.time())  # FIXME: handle timezones
            pipe.zrangebyscore(self.schedule_key, 0, now)
            pipe.zremrangebyscore(self.schedule_key, 0, now)
            events, n = pipe.execute()
            for event in events:
                event = msgpack.loads(event, encoding='utf-8')
                self.emit(event['event_type'], event['payload'])
            gevent.sleep(1)

import gevent
import msgpack
import redis
import time

from lymph.core.interfaces import Interface
from lymph.core.decorators import rpc
from lymph.utils import make_id


class Scheduler(Interface):
    service_type = 'scheduler'
    schedule_key = 'schedule'

    def __init__(self, *args, **kwargs):
        super(Scheduler, self).__init__(*args, **kwargs)
        self.redis = redis.StrictRedis()

    def on_start(self):
        self.container.spawn(self.loop)

    @rpc()
    def schedule(self, channel, eta, event_type, payload):
        self.redis.zadd(self.schedule_key, eta, msgpack.dumps({
            'id': make_id(),
            'event_type': event_type,
            'payload': payload,
        }))
        channel.ack()

    def loop(self):
        while True:
            pipe = self.redis.pipeline()
            now = int(time.time())
            pipe.zrangebyscore(self.schedule_key, 0, now)
            pipe.zremrangebyscore(self.schedule_key, 0, now)
            events, n = pipe.execute()
            for event in events:
                event = msgpack.loads(event, encoding='utf-8')
                self.emit(event['event_type'], event['payload'])
            gevent.sleep(1)

Include ids for scheduled events to make schedule data unique
Include ids for scheduled events to make schedule data unique

  import gevent
  import msgpack
  import redis
  import time
  
  from lymph.core.interfaces import Interface
  from lymph.core.decorators import rpc
+ from lymph.utils import make_id
  
  
  class Scheduler(Interface):
      service_type = 'scheduler'
      schedule_key = 'schedule'
  
      def __init__(self, *args, **kwargs):
          super(Scheduler, self).__init__(*args, **kwargs)
          self.redis = redis.StrictRedis()
  
      def on_start(self):
          self.container.spawn(self.loop)
  
      @rpc()
      def schedule(self, channel, eta, event_type, payload):
          self.redis.zadd(self.schedule_key, eta, msgpack.dumps({
+             'id': make_id(),
              'event_type': event_type,
              'payload': payload,
          }))
          channel.ack()
  
      def loop(self):
          while True:
              pipe = self.redis.pipeline()
-             now = int(time.time())  # FIXME: handle timezones
+             now = int(time.time())
              pipe.zrangebyscore(self.schedule_key, 0, now)
              pipe.zremrangebyscore(self.schedule_key, 0, now)
              events, n = pipe.execute()
              for event in events:
                  event = msgpack.loads(event, encoding='utf-8')
                  self.emit(event['event_type'], event['payload'])
              gevent.sleep(1)
  
Include ids for scheduled events to make schedule data unique
## Code Before:
import gevent
import msgpack
import redis
import time

from lymph.core.interfaces import Interface
from lymph.core.decorators import rpc


class Scheduler(Interface):
    service_type = 'scheduler'
    schedule_key = 'schedule'

    def __init__(self, *args, **kwargs):
        super(Scheduler, self).__init__(*args, **kwargs)
        self.redis = redis.StrictRedis()

    def on_start(self):
        self.container.spawn(self.loop)

    @rpc()
    def schedule(self, channel, eta, event_type, payload):
        self.redis.zadd(self.schedule_key, eta, msgpack.dumps({
            'event_type': event_type,
            'payload': payload,
        }))
        channel.ack()

    def loop(self):
        while True:
            pipe = self.redis.pipeline()
            now = int(time.time())  # FIXME: handle timezones
            pipe.zrangebyscore(self.schedule_key, 0, now)
            pipe.zremrangebyscore(self.schedule_key, 0, now)
            events, n = pipe.execute()
            for event in events:
                event = msgpack.loads(event, encoding='utf-8')
                self.emit(event['event_type'], event['payload'])
            gevent.sleep(1)

## Instruction:
Include ids for scheduled events to make schedule data unique
## Code After:
import gevent
import msgpack
import redis
import time

from lymph.core.interfaces import Interface
from lymph.core.decorators import rpc
from lymph.utils import make_id


class Scheduler(Interface):
    service_type = 'scheduler'
    schedule_key = 'schedule'

    def __init__(self, *args, **kwargs):
        super(Scheduler, self).__init__(*args, **kwargs)
        self.redis = redis.StrictRedis()

    def on_start(self):
        self.container.spawn(self.loop)

    @rpc()
    def schedule(self, channel, eta, event_type, payload):
        self.redis.zadd(self.schedule_key, eta, msgpack.dumps({
            'id': make_id(),
            'event_type': event_type,
            'payload': payload,
        }))
        channel.ack()

    def loop(self):
        while True:
            pipe = self.redis.pipeline()
            now = int(time.time())
            pipe.zrangebyscore(self.schedule_key, 0, now)
            pipe.zremrangebyscore(self.schedule_key, 0, now)
            events, n = pipe.execute()
            for event in events:
                event = msgpack.loads(event, encoding='utf-8')
                self.emit(event['event_type'], event['payload'])
            gevent.sleep(1)
