@Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        if (logger.isDebugEnabled()){
            logger.debug(input.toString());
        }
        this.outputcollector = collector;
        String sourceStreamId = input.getSourceStreamId();
        if (PreconditionBolt.PRECONDITION.equals(sourceStreamId)) {
            String imsi = input.getString(0);
            String eventType = input.getString(1);
            long time = input.getLong(2);
            String lac = input.getString(3);
            String cell = input.getString(4);
            try {
                userGroup.onSignal(time, eventType, imsi, lac, cell);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
        else if (PreconditionBolt.UPDATETIME.equals(sourceStreamId)) {
            long time = input.getLong(0);
            if (time > lastSignalTime){
                userGroup.updateGlobleTime(time, input.getString(1));
                lastSignalTime = time;
                if (logger.isDebugEnabled()){
                    logger.debug(input.toString());
                }
            } else {
                logger.info("drop time:" + time + "/" + TimeUtil.getTime(time) + " < " + lastSignalTime);
            }
        }
    }