def process(self, tup):
        message_buf = tup.values[0]
        message = json.loads(message_buf)

        message_id = message.get('message_id')
        if message_id == PUT_METRIC_DATA_MSG_ID:
            metric_key = str(self.get_metric_key(message))
            storm.emit([metric_key, message_buf])
        elif message_id == PUT_METRIC_ALARM_MSG_ID:
            metric_key = message.get('metric_key')
            storm.emit([metric_key, message_buf])
        elif message_id == DELETE_ALARMS_MSG_ID:
            project_id = message.get('project_id')
            alarmkeys = message.get('alarmkeys')
            for alarmkey in alarmkeys:
                try:
                    alarmkey_uuid = UUID(alarmkey)
                    metric_key = self.get_alarm_metric_key(alarmkey_uuid)
                    metric_key = str(metric_key)
                    if metric_key:
                        message['alarmkey'] = alarmkey
                        storm.emit([metric_key, json.dumps(message)])
                except Exception as e:
                    storm.log("Alarm %s does not exists" % alarmkey)
                    storm.log(traceback.format_exc(e))
        elif message_id == SET_ALARM_STATE_MSG_ID:
            project_id = message.get('project_id')
            alarm_name = message.get('alarm_name')
            alarm_key = self.cass.get_metric_alarm_key(project_id,
                                                       alarm_name)
            if alarm_key:
                alarm = self.cass.get_metric_alarm(alarm_key)
                metric_key = str(alarm.get('metric_key'))
                storm.emit([metric_key, json.dumps(message)])

 def nextTuple(self):
     try:
         input_tuple = None
         try:
             if self.counter >= self.script._num_iterations:
                 return
         except:
             try:
                 input_tuple = self.script._static_input.pop(0)
             except AttributeError:
                 # there is no static input
                 pass
             except IndexError:
                 # static input is empty - no more processing
                 return
         outputs = self.script.process(input_tuple)
         if outputs is None:
             return
         for streamname, output in outputs.iteritems():
             result = output if isinstance(output, list) else [output]
             storm.emit(result, stream=streamname, id=self.counter)
             storm.log("Dispel4Py ------> %s: emitted tuple %s to stream %s" % (self.script.id, result, streamname))
             self.counter += 1
     except:
         # logging the error but it should be passed to client somehow
         storm.log("Dispel4Py ------> %s: %s" % (self.scriptname, traceback.format_exc(), ))

 def process(self, tup):
     url = tup.values[0]
     storm.log("HARing "+url)
     output = self.get_har_with_image(url)
     if output is not "FAIL":
         storm.emit(output, anchors=[tup])
         storm.ack(tup)

 def received_message(self, message):
     uid = str(message)
     if len(uid) and _server:
         log('[OutletWebSocket] Client connected: %s' % uid)
         index = _server.manager.websockets.values().index(self)
         _clients[uid] = _server.manager.websockets.keys()[index]
         emitBolt(['connect', uid], stream='control')

 def process(self, tup):
     word = tup.values[0];
     if (random() < 0.75):
         storm.emit([word + 'lalala'], anchors=[tup])
         storm.ack(tup)
     else:
         storm.log(word + ' randomly skipped!')

    def process_set_alarm_state_msg(self, metric_key, message):
        project_id = message.get('project_id')
        alarm_name = message.get('alarm_name')
        state_reason_data = message.get('state_reason_data')
                      
        if metric_key not in self.metrics:
            self.metrics[metric_key] = MetricMonitor(metric_key, self.cass)

        metric = self.metrics[metric_key]
        
        ret = self.cass.get_metric_alarm_key(project_id, alarm_name)
        if ret:
            alarm_key = ret
            try:
                metricalarm = metric.alarms[alarm_key]
            except KeyError:
                storm.log("alarm key [%s] is found, but alarm is not found." % alarm_key)
                return            
        else:
            storm.log("alarm key [%s] is not found." % alarm_key)
            return
        
        metricalarm['state_reason'] = message.get('state_reason')
        metricalarm['state_value'] = message.get('state_value')
        metricalarm['state_reason_data'] = message.get('state_reason_data')

        # write into database
        alarm_columns = {'state_reason':message.get('state_reason'),
                         'state_value':message.get('state_value')}
        if state_reason_data:
            alarm_columns['state_reason_data'] = state_reason_data
        
        self.cass.put_metric_alarm(alarm_key, alarm_columns)

 def emit(self, record):
     from petrel import storm
     msg = self.format(record)
     for line in msg.split('\n'):
         formatted_line = self.format_string % line
         #print >> sys.stderr, "Calling storm.log with: %s" % formatted_line
         storm.log('%s' % formatted_line)

 def initialize(self, conf, context):
     try:        
         self.counter = 0
         self.modname = conf["dispel4py.module"]
         self.scriptname = conf["dispel4py.script"]
         
         scriptconfig = pickle.loads(str(conf['dispel4py.config'])) if 'dispel4py.config' in conf else {}
         
         storm.log("Dispel4Py ------> loading script %s" % self.scriptname)
         mod = import_module(self.modname)
         self.script = getattr(mod, self.scriptname)()
         for key, value in scriptconfig.iteritems():
             storm.log("Dispel4Py ------> %s: setting attribute %s" % (self.scriptname, key))
             setattr(self.script, key, value)
         storm.log("Dispel4Py ------> loaded script %s" % self.scriptname)
             
         # attach an output writer to each output connection
         for outputname, output in self.script.outputconnections.iteritems():
             output['writer'] = OutputWriter(self.scriptname, outputname)
             
         # pre-processing if required
         self.script.preprocess()
         storm.log("Dispel4Py ------> %s: preprocess() completed." % (self.scriptname,))
     except:
         storm.log("Dispel4Py ------> %s: %s" % (self.scriptname, traceback.format_exc(),))
         raise

 def fail(self, msg_id):
     tup, retries = self.buffer[msg_id]
     if retries >= 5:
         del self.buffer[msg_id]
         log('[RabbitMQSpout] Message %s failed for good.' % msg_id)
     else:
         self.buffer[msg_id] = (tup, retries + 1)
         emit(tup, id=msg_id)

 def fail(self, cnt_id):
     tup, retries = self.buffer[cnt_id]
     if retries >= 5:
         del self.buffer[cnt_id]
         log('[ZonAPISpout] Message %s failed for good.' % cnt_id)
     else:
         self.buffer[cnt_id] = (tup, retries + 1)
         emit(tup, id=cnt_id)

 def nextTuple(self):
     if self.count < 2:
         word = choice(words)
         id = str(uuid4())
         self.pending[id] = word
         storm.rpcMetrics("my-custom-shellspout-metric", 1)
         self.count = self.count + 1
         storm.log("TesterSpout update my-custom-shellspout-metric "+str(self.count))
         storm.emit([word], id=id)

def stormLogger(message):
    ''' 
    Logger implementation for Storm.
    '''
    try:
        storm.log(str(message))
    except:
        # just ignore the exception
        pass

def log_tweeter_error(tweep_error, sleep_time=2):
    '''
    :param tweep_error: Exception dealing with twitter to log to the parent process
    :type api: tweepy.TweepError

    :param sleep_time: time in seconds to sleep before continuing the execution
    '''
    # We have hit the REST API Rate limit for Twitter https://dev.twitter.com/docs/rate-limiting/1.1, no more tweets for some time
    storm.log("Tweepy error {error}, sleeping for {secs} seconds in case Twitter rate limit has been hit".format(error=str(tweep_error), secs=sleep_time))
    time.sleep(sleep_time)

 def process(self, tup):
     word = tup.values[0]
     if self.counts.has_key(word):
         count = self.counts[word]
     else:
         count = 0
     count += 1
     self.counts[word] = count
     storm.log(str(word)+" "+str(count))
     storm.emit([word, count])

 def write(self, output):
     result = output if isinstance(output, list) else [output]
     try:
         storm.emit(result, stream=self.streamname)
         storm.log("Dispel4Py ------> Emitted to stream %s." % (self.scriptname, self.streamname))
     except TypeError:
         # encode manually
         encoded = encode_types(result)
         storm.emit(encoded, stream=self.streamname)
         storm.log("Dispel4Py ------> Emitted to stream %s." % (self.scriptname, self.streamname))

    def nextTuple(self):

        try:
            vb, msg = self.reader.response().next()
            if msg and msg['opcode'] == 87:
                tweet = json.loads(msg['value'])
                tags = tweet['hashtags']
                id = tweet['id']
                storm.emit([tags[0], id, vb])
        except StopIteration:
            storm.log("resetting vbucket map")
            self.reader.reset()

  def process(self, tuple):
    storm.log(tuple.component + ': ' + str(tuple.values))

    # Write rules to local fs and reload
    if tuple.component == 'rule-definitions':
      fn = tuple.values[0].split('/')[-1]
      self.write_rule(fn, tuple.values[1])
      self.define_rule(fn)
    # otherwise, apply each rule to each tuple
    else:
      for module in self.modules:
        output = self.modules[module].handle(tuple)
        storm.emit([output])

    def initialize(self, conf, context):
        host = conf.get('zeit.recommend.elasticsearch.host', 'localhost')
        port = conf.get('zeit.recommend.elasticsearch.port', 9200)
        self.es = Elasticsearch(hosts=[{'host': host, 'port': port}])
        self.match = re.compile('seite-[0-9]|komplettansicht').match
        self.index = '%s-%s' % date.today().isocalendar()[:2]
        ic = IndicesClient(self.es)

        try:
            if not ic.exists(self.index):
                ic.create(self.index)
        except ConnectionError, e:
            log('[UserIndexBolt] ConnectionError, index unreachable: %s' % e)
            return

 def get_stats(tmp_stat):
     try:
         ret = dict(zip(self.cass.STATISTICS,
                             map(lambda x: x.values()[0], tmp_stat)))
         for v in ret:
             if v == None: v = float('nan') 
     except IndexError:
         storm.log("index %s is not in DB." % time_idx)
         ret = {'SampleCount' : float('nan'),
                 'Sum' : float('nan'),
                 'Average' : float('nan'),
                 'Minimum' : float('nan'),
                 'Maximum' : float('nan') }
     return ret

 def process(self,tup):
     # storm.log("Dispel4Py ------> %s: Received block." % (self.script.id, ))
     try:
         inputname = self.inputmapping[tup.component][tup.stream]
         storm.log("Dispel4Py ------> %s: Received block at input '%s'" % (self.script.id, inputname, ))
         # inputs = tup.values
         inputs = decode_types(tup.values)
         outputs = self.script.process( { inputname : inputs })
         # storm.log("Dispel4Py ------> %s: Processing complete." % self.scriptname)
         
         if outputs is None:
             return
         for streamname, output in outputs.iteritems():
             result = output if isinstance(output, list) else [output]
             try:
                 storm.emit(result, stream=streamname)
                 storm.log("Dispel4Py ------> %s: Emitted to stream %s: %s" % (self.script.id, streamname, str(result)[:200]))
             except TypeError:
                 # encode manually
                 encoded = encode_types(result)
                 storm.emit(encoded, stream=streamname)
                 storm.log("Dispel4Py ------> %s: Emitted to stream %s" % (self.script.id, streamname))
             # except:
             #     storm.log("%s: %s" % (self.script.id, traceback.format_exc()))
     except:
         storm.log("Dispel4Py ------> %s: %s" % (self.script.id, traceback.format_exc(), ))