/* Helper function to free the context. */
static void __redisAsyncFree(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    redisCallback cb;
    dictIterator *it;
    dictEntry *de;

    /* Execute pending callbacks with NULL reply. */
    while (__redisShiftCallback(&ac->replies,&cb) == REDIS_OK)
        __redisRunCallback(ac,&cb,NULL);

    /* Execute callbacks for invalid commands */
    while (__redisShiftCallback(&ac->sub.invalid,&cb) == REDIS_OK)
        __redisRunCallback(ac,&cb,NULL);

    /* Run subscription callbacks callbacks with NULL reply */
    it = dictGetIterator(ac->sub.channels);
    while ((de = dictNext(it)) != NULL)
        __redisRunCallback(ac,dictGetEntryVal(de),NULL);
    dictReleaseIterator(it);
    dictRelease(ac->sub.channels);

    it = dictGetIterator(ac->sub.patterns);
    while ((de = dictNext(it)) != NULL)
        __redisRunCallback(ac,dictGetEntryVal(de),NULL);
    dictReleaseIterator(it);
    dictRelease(ac->sub.patterns);

    /* Signal event lib to clean up */
    _EL_CLEANUP(ac);

    /* Execute disconnect callback. When redisAsyncFree() initiated destroying
     * this context, the status will always be REDIS_OK. */
    if (ac->onDisconnect && (c->flags & REDIS_CONNECTED)) {
        if (c->flags & REDIS_FREEING) {
            ac->onDisconnect(ac,REDIS_OK);
        } else {
            ac->onDisconnect(ac,(ac->err == 0) ? REDIS_OK : REDIS_ERR);
        }
    }

    /* Cleanup self */
    redisFree(c);
}

void movekeysCommand(redisClient *c) {
    redisDb *src, *dst;
    int srcid;

    dictIterator *di;
    dictEntry *de;
    sds pattern = c->argv[1]->ptr;
    int plen = sdslen(pattern), allkeys;
    unsigned long numkeys = 0;

    /* Obtain source and target DB pointers */
    src = c->db;
    srcid = c->db->id;
    if (selectDb(c,atoi(c->argv[2]->ptr)) == REDIS_ERR) {
        addReply(c,shared.outofrangeerr);
        return;
    }
    dst = c->db;
    selectDb(c,srcid); /* Back to the source DB */

    /* If the user is moving using as target the same
     * DB as the source DB it is probably an error. */
    if (src == dst) {
        addReply(c,shared.sameobjecterr);
        return;
    }

    di = dictGetIterator(c->db->dict);
    allkeys = (pattern[0] == '*' && pattern[1] == '\0');
    while((de = dictNext(di)) != NULL) {
        sds key = dictGetEntryKey(de);
        robj *keyobj;

        if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) {
            keyobj = createStringObject(key,sdslen(key));
            if (expireIfNeeded(c->db,keyobj) == 0) {
                robj *val = dictGetEntryVal(de);

                /* Try to add the element to the target DB */
                if (dbAdd(dst,keyobj,val) != REDIS_ERR) {
                    incrRefCount(val);

                    /* OK! key moved, free the entry in the source DB */
                    dbDelete(src,keyobj);
                    server.dirty++;
                    numkeys++;
                }
            }
            decrRefCount(keyobj);
        }
    }
    dictReleaseIterator(di);
    addReplyLongLong(c,numkeys);
}

/* Move to the next entry in the set. Returns the object at the current
 * position.
 *
 * Since set elements can be internally be stored as redis objects or
 * simple arrays of integers, setTypeNext returns the encoding of the
 * set object you are iterating, and will populate the appropriate pointer
 * (eobj) or (llobj) accordingly.
 *
 * When there are no longer elements -1 is returned.
 * Returned objects ref count is not incremented, so this function is
 * copy on write friendly. */
int setTypeNext(setTypeIterator *si, robj **objele, int64_t *llele) {
    if (si->encoding == REDIS_ENCODING_HT) {
        dictEntry *de = dictNext(si->di);
        if (de == NULL) return -1;
        *objele = dictGetEntryKey(de);
    } else if (si->encoding == REDIS_ENCODING_INTSET) {
        if (!intsetGet(si->subject->ptr,si->ii++,llele))
            return -1;
    }
    return si->encoding;
}

void  destroyTaskMap(dict* p) 
{
    if (p) {
        dictIterator* di = dictGetIterator(p);
        dictEntry* de;
        while ((de = dictNext(di)) != NULL) {
            ugTaskType* ptask = dictGetEntryVal(de); 
            luaworkUnrefFunction(server.ls, ptask->handle, NULL);            
        }
        dictReleaseIterator(di);
        dictRelease(p);
    }
}

/* Unsubscribe from all the channels. Return the number of channels the
 * client was subscribed from. */
int pubsubUnsubscribeAllChannels(redisClient *c, int notify) {
    dictIterator *di = dictGetSafeIterator(c->pubsub_channels);
    dictEntry *de;
    int count = 0;

    while((de = dictNext(di)) != NULL) {
        robj *channel = dictGetKey(de);

        count += pubsubUnsubscribeChannel(c,channel,notify);
    }
    dictReleaseIterator(di);
    return count;
}

void
_recurse_each(Callback *cb, dict *hash) {
	dictIterator * iterator = dictGetIterator(hash);
	dictEntry *next;
	while((next = dictNext(iterator))) {
		QueryNode *node = dictGetVal(next);
		cb->handler(cb, node);
		if(node->children) {
			_recurse_each(cb, node->children);
		}
	}
  dictReleaseIterator(iterator);
}

static void shutdown_server(EV_P_ ev_signal *w, int revents) {
  IGNORE_VAR(w); IGNORE_VAR(revents);
  IGNORE_VAR(dictGenHashFunction); IGNORE_VAR(dictCreate); IGNORE_VAR(dictReplace); IGNORE_VAR(dictRelease);
  dictEntry *de;
  dictIterator *iter = dictGetIterator(active_connections);
  while ((de = dictNext(iter)) != NULL) {
    muxConn *mc = de->val;
    // stop any pending writes
    ev_io_stop(EV_A_ mc->watcher);
    disconnectAndClean(mc);
  }
  dictReleaseIterator(iter);
  ev_break(EV_A_ EVBREAK_ALL);
}

/* Move to the next entry in the set. Returns the object at the current
 * position.
 *
 * Since set elements can be internally be stored as redis objects or
 * simple arrays of integers, setTypeNext returns the encoding of the
 * set object you are iterating, and will populate the appropriate pointer
 * (objele) or (llele) accordingly.
 *
 * Note that both the objele and llele pointers should be passed and cannot
 * be NULL since the function will try to defensively populate the non
 * used field with values which are easy to trap if misused.
 *
 * When there are no longer elements -1 is returned.
 * Returned objects ref count is not incremented, so this function is
 * copy on write friendly. */
int setTypeNext(setTypeIterator *si, robj **objele, int64_t *llele) {
    if (si->encoding == REDIS_ENCODING_HT) {
        dictEntry *de = dictNext(si->di);
        if (de == NULL) return -1;
        *objele = dictGetKey(de);
        *llele = -123456789; /* Not needed. Defensive. */
    } else if (si->encoding == REDIS_ENCODING_INTSET) {
        if (!intsetGet(si->subject->ptr,si->ii++,llele))
            return -1;
        *objele = NULL; /* Not needed. Defensive. */
    } else {
        redisPanic("Wrong set encoding in setTypeNext");
    }
    return si->encoding;
}

void
slotsmgrt_cleanup() {
    dictIterator *di = dictGetSafeIterator(server.slotsmgrt_cached_sockfds);
    dictEntry *de;
    while((de = dictNext(di)) != NULL) {
        slotsmgrt_sockfd *pfd = dictGetVal(de);
        if ((server.unixtime - pfd->lasttime) > 30) {
            redisLog(REDIS_WARNING, "slotsmgrt: timeout target %s, lasttime = %ld, now = %ld",
                   (char *)dictGetKey(de), pfd->lasttime, server.unixtime);
            dictDelete(server.slotsmgrt_cached_sockfds, dictGetKey(de));
            close(pfd->fd);
            zfree(pfd);
        }
    }
    dictReleaseIterator(di);
}

void expirekeysGenericCommand(redisClient *c, robj *keypattern, robj *param, long offset) {
    dictIterator *di;
    dictEntry *de;
    sds pattern = keypattern->ptr;
    int plen = sdslen(pattern);
    unsigned long numkeys = 0, allkeys;
    time_t seconds;
    time_t when;

    if (getLongFromObjectOrReply(c, param, &seconds, NULL) != REDIS_OK) return;

    seconds -= offset;
    when = time(NULL)+seconds;

    di = dictGetIterator(c->db->dict);
    allkeys = (pattern[0] == '*' && pattern[1] == '\0');
    while((de = dictNext(di)) != NULL) {
        sds key = dictGetEntryKey(de);
        robj *keyobj;

        if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) {
            keyobj = createStringObject(key,sdslen(key));
            if (seconds <= 0 && !server.loading && !server.masterhost) {
                robj *aux;

                redisAssert(dbDelete(c->db,keyobj));
                server.dirty++;
                numkeys++;

                /* Replicate/AOF this as an explicit DEL. */
                aux = createStringObject("DEL",3);
                rewriteClientCommandVector(c,2,aux,keyobj);
                decrRefCount(aux);
                touchWatchedKey(c->db,keyobj);
            } else {
                time_t when = time(NULL)+seconds;
                setExpire(c->db,keyobj,when);
                touchWatchedKey(c->db,keyobj);
                server.dirty++;
                numkeys++;
            }
            decrRefCount(keyobj);
        }
    }
    dictReleaseIterator(di);
    addReplyLongLong(c,numkeys);
}

void call_bridge_event(struct redisClient *c,int triggle_place,int event_type)
{

    if(event_type==-1)
    {
        sds cmds=(sds)sdsdup(c->argv[0]->ptr);
        sdstolower(cmds);

        if(triggle_place==TRIGGLE_BEFORE) //no event given
        {
            cmds=sdscat(cmds,"ing");


        }
        if(triggle_place==TRIGGLE_AFTER)
        {
            cmds=sdscat(cmds,"ed");

        }

        event_type=process_trigglecmd(cmds);

  //      redisLog(REDIS_NOTICE,"cmds:%s id:%d",cmds,event_type);
        sdsfree(cmds);
    }

    if(event_type==-1)
        return;
    struct dictIterator *iter=dictGetIterator(server.bridge_db.triggle_scipts[c->db->id]);
    dictEntry *trigs;
    do{
        trigs=dictNext(iter);
        if(trigs!=NULL)
        {
            struct bridge_db_triggle_t * tmptrg=dictGetVal(trigs);

            if(tmptrg->event==event_type&&strncmp(c->argv[1]->ptr,dictGetKey(trigs),sdslen(dictGetKey(trigs)))==0){ //找到指定的类型事件
            //if(tmptrg->event==event_type){ //找到指定的类型事件
//                redisLog(REDIS_NOTICE,"triggle_event:%d,%s",event_type,(char *)dictGetKey(trigs));
                triggle_event(c,dictGetKey(trigs));
            }
        }
    }while(trigs!=NULL);
    dictReleaseIterator(iter);


}

/* Reset data for the specified event, or all the events data if 'event' is
 * NULL.
 *
 * Note: this is O(N) even when event_to_reset is not NULL because makes
 * the code simpler and we have a small fixed max number of events. */
int latencyResetEvent(char *event_to_reset) {
    dictIterator *di;
    dictEntry *de;
    int resets = 0;

    di = dictGetSafeIterator(server.latency_events);
    while((de = dictNext(di)) != NULL) {
        char *event = dictGetKey(de);

        if (event_to_reset == NULL || strcasecmp(event,event_to_reset) == 0) {
            dictDelete(server.latency_events, event);
            resets++;
        }
    }
    dictReleaseIterator(di);
    return resets;
}

void rdb_save_triggles(rio *rdb)
{

    //save event 
    //db_num int int int int 
    //db 
    //scripts_num
    //key event lua_scripts 
    //key event lua_scripts 
    //.......
    dictIterator *di = NULL;
    dictEntry *de;   
    int i=0;
    for(i=0;i<server.dbnum;i++){
        int eventid=server.bridge_db.bridge_event[i]; 
        rioWrite(rdb,&eventid,4);
    }
    for(i=0;i<server.dbnum;i++)
    {
        dict *d = server.bridge_db.triggle_scipts[i];
        int mysize=dictSize(d);
        rioWrite(rdb,&mysize,4);
        if (dictSize(d) == 0) continue;
        di = dictGetSafeIterator(d);
        if (!di) {
            return ;
        }
        /* Iterate this DB writing every entry */
        while((de = dictNext(di)) != NULL) {
            sds keystr = dictGetKey(de);
            robj key;
            initStaticStringObject(key,keystr);
            if (rdbSaveStringObject(rdb,&key) == -1) return;

            struct bridge_db_triggle_t * tmptrg=dictGetVal(de);
            int event_id=tmptrg->event; 
            rioWrite(rdb,&event_id,4);
            int db_id=tmptrg->dbid; 
            rioWrite(rdb,&db_id,4);
            if (rdbSaveObjectType(rdb,tmptrg->lua_scripts) == -1) return ;
            if (rdbSaveObject(rdb,tmptrg->lua_scripts) == -1) return ; 
        }
    }
    if (di) dictReleaseIterator(di);
}

/* PUBSUB command for Pub/Sub introspection. */
void pubsubCommand(client *c) {
    if (!strcasecmp(c->argv[1]->ptr,"channels") &&
        (c->argc == 2 || c->argc ==3))
    {
        /* PUBSUB CHANNELS [<pattern>] */
        sds pat = (c->argc == 2) ? NULL : c->argv[2]->ptr;
        dictIterator *di = dictGetIterator(server.pubsub_channels);
        dictEntry *de;
        long mblen = 0;
        void *replylen;

        replylen = addDeferredMultiBulkLength(c);
        while((de = dictNext(di)) != NULL) {
            robj *cobj = dictGetKey(de);
            sds channel = cobj->ptr;

            if (!pat || stringmatchlen(pat, sdslen(pat),
                                       channel, sdslen(channel),0))
            {
                addReplyBulk(c,cobj);
                mblen++;
            }
        }
        dictReleaseIterator(di);
        setDeferredMultiBulkLength(c,replylen,mblen);
    } else if (!strcasecmp(c->argv[1]->ptr,"numsub") && c->argc >= 2) {
        /* PUBSUB NUMSUB [Channel_1 ... Channel_N] */
        int j;

        addReplyMultiBulkLen(c,(c->argc-2)*2);
        for (j = 2; j < c->argc; j++) {
            list *l = dictFetchValue(server.pubsub_channels,c->argv[j]);

            addReplyBulk(c,c->argv[j]);
            addReplyLongLong(c,l ? listLength(l) : 0);
        }
    } else if (!strcasecmp(c->argv[1]->ptr,"numpat") && c->argc == 2) {
        /* PUBSUB NUMPAT */
        addReplyLongLong(c,listLength(server.pubsub_patterns));
    } else {
        addReplyErrorFormat(c,
            "Unknown PUBSUB subcommand or wrong number of arguments for '%s'",
            (char*)c->argv[1]->ptr);
    }
}

/* latencyCommand() helper to produce the reply for the LATEST subcommand,
 * listing the last latency sample for every event type registered so far. */
void latencyCommandReplyWithLatestEvents(client *c) {
    dictIterator *di;
    dictEntry *de;

    addReplyMultiBulkLen(c,dictSize(server.latency_events));
    di = dictGetIterator(server.latency_events);
    while((de = dictNext(di)) != NULL) {
        char *event = dictGetKey(de);
        struct latencyTimeSeries *ts = dictGetVal(de);
        int last = (ts->idx + LATENCY_TS_LEN - 1) % LATENCY_TS_LEN;

        addReplyMultiBulkLen(c,4);
        addReplyBulkCString(c,event);
        addReplyLongLong(c,ts->samples[last].time);
        addReplyLongLong(c,ts->samples[last].latency);
        addReplyLongLong(c,ts->max);
    }
    dictReleaseIterator(di);
}

sds _masterGetStatus() {
    /*TODO: calculate cache increase speed,
     * then adopt a suitable stale-cache freeing strategy
     * Three involved params:
     * (1) master sleep,
     * (2) ae loop wait,
     * and (3) number of stale entries in one ae loop
     */

    sds status = sdsempty();//sdsfromlonglong(master_total_mem);
    status = sdscatprintf(status,"TOL RAM: %-6.2lfMB\tUSED RAM: %-6.2lf\n",
                          BYTES_TO_MEGABYTES(MASTER_MAX_AVAIL_MEM),
                          BYTES_TO_MEGABYTES(master_total_mem));
#if (CCACHE_LOG_LEVEL == CCACHE_DEBUG)
    status = sdscatprintf(status,"Detail:\n");
    status = sdscatprintf(status,"%-3s %-32s: %-6s\n"," ","KEY","MEM");
    dictIterator *di = dictGetIterator(master_cache);
    dictEntry *de;
    int idx = 1;
    while((de = dictNext(di)) != NULL) {
        objSds *value = (objSds*)dictGetEntryVal(de);
        if(value) {
            if(value->ptr) {
                status = sdscatprintf(status,"%-3d %-32s: %-6ld\n",
                                        idx++,
                                        (char*)dictGetEntryKey(de),
                                        sdslen(value->ptr));
            }
            else {
                status = sdscatprintf(status,"%-3d %-32s: %-6s\n",
                                        idx++,
                                        (char*)dictGetEntryKey(de),
                                        "WAITING");
            }
        }
    }
    dictReleaseIterator(di);
#endif
    sds status_reply = sdsnew("HTTP/1.1 200 OK\r\n");
    status_reply = sdscatprintf(status_reply,"Content-Length: %ld\r\n\r\n%s",sdslen(status),status);
    sdsfree(status);
    return status_reply;
}

/* COMMAND <subcommand> <args> */
void commandCommand(client *c) {
    dictIterator *di;
    dictEntry *de;

    if (c->argc == 1) {
        addReplyMultiBulkLen(c, dictSize(server.commands));
        di = dictGetIterator(server.commands);
        while ((de = dictNext(di)) != NULL) {
            addReplyCommand(c, dictGetVal(de));
        }
        dictReleaseIterator(di);
    } else if (!strcasecmp(c->argv[1]->ptr, "info")) {
        int i;
        addReplyMultiBulkLen(c, c->argc-2);
        for (i = 2; i < c->argc; i++) {
            addReplyCommand(c, dictFetchValue(server.commands, c->argv[i]->ptr));
        }
    } else if (!strcasecmp(c->argv[1]->ptr, "count") && c->argc == 2) {
        addReplyLongLong(c, dictSize(server.commands));
    } else if (!strcasecmp(c->argv[1]->ptr,"getkeys") && c->argc >= 3) {
        struct redisCommand *cmd = lookupCommand(c->argv[2]->ptr);
        int *keys, numkeys, j;

        if (!cmd) {
            addReplyErrorFormat(c,"Invalid command specified");
            return;
        } else if ((cmd->arity > 0 && cmd->arity != c->argc-2) ||
                   ((c->argc-2) < -cmd->arity))
        {
            addReplyError(c,"Invalid number of arguments specified for command");
            return;
        }

        keys = getKeysFromCommand(cmd,c->argv+2,c->argc-2,&numkeys);
        addReplyMultiBulkLen(c,numkeys);
        for (j = 0; j < numkeys; j++) addReplyBulk(c,c->argv[keys[j]+2]);
        getKeysFreeResult(keys);
    } else {
        addReplyError(c, "Unknown subcommand or wrong number of arguments.");
        return;
    }
}

dictEntry *roZDictNext(roZDictIter *iter) {
    dictEntry *de = NULL;

    cowLock();
    if (iter->ar != NULL) {
        if (iter->pos >= 0 && iter->pos < iter->ar->numele) {
            de = &iter->ar->zde[iter->pos].de;
            iter->pos++;
        }
    } else if (iter->di != NULL) {
        de = dictNext(iter->di);
        iter->pos++;
    }
    if (de == NULL) {
        iter->pos = -1;
    }
    cowUnlock();

    return de;
}

/* convert a hash dictionary encoding to a dictionary array encoding */
cowDictZArray *cowConvertDictToZArray(dict *hdict) {
    dictIterator * di;
    dictEntry *de;
    int dsize;
    cowDictZArray *dar;
    int dcount = 0;
    dictZEntry *dezNew;
    dictZEntry *dezPrev;

    /* create copy */
    dsize = dictSize(hdict) > dictSlots(hdict) ? dictSize(hdict) : dictSlots(hdict);
    dar = (cowDictZArray *)zmalloc(sizeof(cowDictZArray) +
                                (dsize * sizeof(dictZEntry)) );

    /* copy all entries without refcounting or copying values */
    /* can't just memcpy the whole dictionary because entries are allocated */
    di = dictGetSafeIterator(hdict);
    dezNew = &dar->zde[0];
    dezPrev = NULL;
    while((de = dictNext(di)) != NULL && dcount < dsize) {
        double *score = (double *)dictGetEntryVal(de);
        /* copy score value into array
            and point val to score.  */
        dezNew->de.key = de->key;
        dezNew->score = *score;
        dezNew->de.val = &dezNew->score;
        /* fix next ptr of prev entry */
        if (dezPrev != NULL) {
            dezPrev->de.next = &dezNew->de;
        }
        dezPrev = dezNew;
        dezNew++;
        dcount++;
    }
    if (dezPrev != NULL) {
        dezPrev->de.next = NULL;
    }
    dar->numele = dcount;
    dictReleaseIterator(di);
    return dar;
}

/* Unsubscribe from all the channels. Return the number of channels the
 * client was subscribed to. */
int smempubsubUnsubscribeAllChannels(client *c, int notify) 
{
    dictIterator *di = dictGetSafeIterator(c->smempubsub_channels);
    dictEntry *de;
    int count = 0;

    while((de = dictNext(di)) != NULL) {
        robj *channel = dictGetKey(de);

        count += smempubsubUnsubscribeChannel(c,channel,notify);
    }
    /* We were subscribed to nothing? Still reply to the client. */
    if (notify && count == 0) {
        addReply(c,shared.mbulkhdr[3]);
        addReply(c,shared.unsubscribebulk);
        addReply(c,shared.nullbulk);
        addReplyLongLong(c,dictSize(c->smempubsub_channels));
    }
    dictReleaseIterator(di);
    return count;
}