bool RedisManager::connect()
{
  pubContext = redisConnect(config->host, config->port);
  if (pubContext != NULL && pubContext->err)
  {
    printf("%d\n", pubContext->err);
    return false;
  }

  subContext = redisConnect(config->host, config->port);
  if (subContext != NULL && subContext->err)
  {
    printf("%d\n", subContext->err);
    return false;
  }

  successContext = redisConnect(config->host, config->port);
  if(successContext != NULL && successContext->err)
  {
	  printf("%d\n", successContext->err);
	  return false;
  }
  redisCommand(successContext, "set successCount 0");

  totalContext = redisConnect(config->host, config->port);
  if(totalContext != NULL && totalContext->err)
  {
	  printf("%d\n", totalContext->err);
	  return false;
  }
  redisCommand(totalContext, "set totalCount 0");
  return true;
}

static gboolean
redis_dd_connect(RedisDriver *self, gboolean reconnect)
{
  if (reconnect && (self->c != NULL))
    {
      redisCommand(self->c, "ping");

      if (!self->c->err)
        return TRUE;
      else
        self->c = redisConnect(self->host, self->port);
    }
  else
    self->c = redisConnect(self->host, self->port);

  if (self->c->err)
    {
      msg_error("REDIS server error, suspending",
                evt_tag_str("driver", self->super.super.super.id),
                evt_tag_str("error", self->c->errstr),
                evt_tag_int("time_reopen", self->super.time_reopen),
                NULL);
      return FALSE;
    }
  else
    msg_debug("Connecting to REDIS succeeded",
              evt_tag_str("driver", self->super.super.super.id), NULL);

  return TRUE;
}

int redisPoolInit(redisPool *pool, char hostname[255], int poolsize)  {

    int i, fd;
    char currentdb[2];
    redisReply *reply;

    pool->size = poolsize;
    reply = redisConnect(&fd, hostname, cfg.redis_port);

    if(reply != NULL) {
        return -1;
    }

    reply = redisCommand(fd, "GET CURRENTDB");

    if(reply->type == REDIS_REPLY_STRING) {
        snprintf(currentdb, strlen(reply->reply) +1, "%s", reply->reply);
    }

    if(atoi(currentdb) == 0) {
        printf("Current db not configured\n");
        syslog(LOG_ERR, "Current db not configured");
        return -1;
    }
        
    syslog(LOG_INFO, "Current database selected: %s", reply->reply);

    freeReplyObject(reply);
    close(fd);

    for(i = 0; i < poolsize; i++) {
        if(pool->fd[i])
           close(pool->fd[i]);

        reply = redisConnect(&pool->fd[i], hostname, cfg.redis_port);
        if(reply != NULL) {
            freeReplyObject(reply);
            return -1;
        }
    
        reply = redisCommand(pool->fd[i], "SELECT %s", currentdb);
        freeReplyObject(reply);
    }

    if(cfg.expire_seconds == 0) {
        reply = redisCommand(pool->fd[0], "KEYS *");
        if(reply->type == REDIS_REPLY_ARRAY)  {
            if(reply->elements == 0 ) {
                printf("Empty database, i'm fuc**ng off\n");
                syslog(LOG_ERR, "Empty database, i'm fuc**ng off");
                freeReplyObject(reply);
                return -1;
            }
        }
        freeReplyObject(reply);
    }
    pool->current = 0;
    return 0;
}

int main(void) {
    int ppid = fork();
    if (ppid == 0) {
        redisContext *c;
        redisReply *reply;
        c = redisConnect((char*)"127.0.0.1", 6379);
        if (c->err) {
            printf("Connection error: %s\n", c->errstr);
            exit(1);
        }
        
        printf("Subscribing\n");
        reply = redisCommand(c,"SUBSCRIBE foo");
        freeReplyObject(reply);
        
        // Should get bar
        redisGetReply(c, (void**)&reply);
        printf("Reply is: %s\n", reply->element[2]->str);
        freeReplyObject(reply);
        
        printf("Unsubscribing\n");
        reply = redisCommand(c,"PUNSUBSCRIBE *");
        printf("Unsubscribed from %s\n", reply->element[1]->str);
        freeReplyObject(reply);
        
        // Really, we should hang here, but we'll get zip
        redisGetReply(c, (void**)&reply);
        printf("Reply is: %s\n", reply->element[2]->str);
        freeReplyObject(reply);
        exit(0);
    }
    else {
        redisContext *c;
        redisReply *reply;
        c = redisConnect((char*)"127.0.0.1", 6379);
        if (c->err) {
            printf("Connection error: %s\n", c->errstr);
            exit(1);
        }
        
        // I expect this to go through
        sleep(1);
        printf("Publishing");
        reply = redisCommand(c, "PUBLISH foo bar");
        freeReplyObject(reply);
        
        // This should come in after the unsubscribe (monitor can confirm this)
        sleep(1);
        printf("Publishing");
        reply = redisCommand(c, "PUBLISH foo zip");
        freeReplyObject(reply);
    }
    wait(0);
    return 0;
}

/* Connect to the client. If force is not zero the connection is performed
 * even if there is already a connected socket. */
static int cliConnect(int force) {
    if (context == NULL || force) {
        if (context != NULL)
            redisFree(context);

        if (config.hostsocket == NULL) {
            context = redisConnect(config.hostip,config.hostport);
        } else {
            context = redisConnectUnix(config.hostsocket);
        }

        if (context->err) {
            fprintf(stderr,"Could not connect to Redis at ");
            if (config.hostsocket == NULL)
                fprintf(stderr,"%s:%d: %s\n",config.hostip,config.hostport,context->errstr);
            else
                fprintf(stderr,"%s: %s\n",config.hostsocket,context->errstr);
            redisFree(context);
            context = NULL;
            return REDIS_ERR;
        }

        /* Do AUTH and select the right DB. */
        if (cliAuth() != REDIS_OK)
            return REDIS_ERR;
        if (cliSelect() != REDIS_OK)
            return REDIS_ERR;
    }
    return REDIS_OK;
}

int main(int argc, char *argv[]) {
  redisReply *reply;
  int deleted_keys_count = 0;

  if (argc < 2) {
    printf("Usage: redis-checksum directory [--dry-run]\n");
    return 1;
  }

  if (argc > 2 && (strcmp(argv[2], "--dry-run") == 0)) {
    dry_run = 1;
  }


  redis = redisConnect("127.0.0.1", 6379);

  if(redis->err) {
    fprintf(stderr, "Connection error: %s\n", redis->errstr);
    exit(EXIT_FAILURE);
  }

  // Get rid of non-existent files from redis
  remove_stale_redis_keys(redis);

  ftw(argv[1], &redis_upsert_file, 10);

  printf("Summary%s:\n", (dry_run) ? " (DRY RUN)" : "");
  printf("Keys Added:   %d\n", counts[REDIS_COUNT_ADDED]);
  printf("Keys Updated: %d\n", counts[REDIS_COUNT_UPDATED]);
  printf("Keys Deleted: %d\n", counts[REDIS_COUNT_DELETED]);

  return 0;
}

/**
 * @brief save_img_db Choose db mode to save images.
 *
 * @param thr_arg Thread arg struct.
 * @param cache_key The key of image.
 * @param buff Image buffer.
 * @param len Image size.
 *
 * @return 1 for succ or -1 for fialed.
 */
int save_img_db(thr_arg_t *thr_arg, const char *cache_key, const char *buff, const size_t len)
{
    int ret = -1;
    if(settings.mode == 2)
        ret = save_img_beansdb(thr_arg->beansdb_conn, cache_key, buff, len);
    else if(settings.mode == 3)
        ret = save_img_ssdb(thr_arg->ssdb_conn, cache_key, buff, len);

    if(ret == -1 && settings.mode == 3)
    {
        if(thr_arg->ssdb_conn != NULL)
            redisFree(thr_arg->ssdb_conn);
        redisContext* c = redisConnect(settings.ssdb_ip, settings.ssdb_port);
        if(c->err)
        {
            redisFree(c);
            LOG_PRINT(LOG_DEBUG, "Connect to ssdb server faile");
            return ret;
        }
        else
        {
            thr_arg->ssdb_conn = c;
            LOG_PRINT(LOG_DEBUG, "SSDB Server Reconnected.");
            ret = save_img_ssdb(thr_arg->ssdb_conn, cache_key, buff, len);
            //evthr_set_aux(thr_arg->thread, thr_arg);
        }
    }
    return ret;
}

redisContext *_myredisConnect(struct config config) {
    redisContext *c = NULL;
    redisReply *reply;
    char cmd[256];
    int len;

   if (config.type == CONN_TCP) {
        c = redisConnect(config.tcp.host, config.tcp.port);
    } else if (config.type == CONN_UNIX_SOCK) {
        c = redisConnectUnix(config.unix_sock.path);
    } else {
        assert(NULL);
    }

    if (c->err && pFile) {
    	info_print("Connection error: %s\n", c->errstr);
	return c;
    }


    /* Authenticate */
    if (config.auth) {
        reply = redisCommand(c,"AUTH %s",config.password);
        if(reply) {
            freeReplyObject(reply);
	}
    }
    return c;
}

bool CRedisUtil::Connect(const char* pszAddr, int nPort, int nDbId)
{
	m_redis = redisConnect(pszAddr, nPort);
	if (m_redis->err != 0)
	{
		LogError("redis_connect", "failed to connect: %s:%d ,err=%s", pszAddr, nPort, m_redis->errstr);
		return false;
	}
	//选择库
	char szCommond[24];
	memset(szCommond, 0, sizeof(szCommond));
	snprintf(szCommond, sizeof(szCommond), "SELECT %d", nDbId);
	redisReply* reply = (redisReply*)redisCommand(m_redis, szCommond);
	if (!reply)
	{
		LogError("redis_connect", szCommond);
		return false;
	}
	else
	{
		LogInfo("redis_connect", szCommond);
		//LogDebug("freeReplyObject", "%s : %d", __FILE__, __LINE__);
		freeReplyObject(reply);
	}
	return true;
}

int send_commands(std::vector<std::string> cmds, std::vector<std::string> times, Json::Value &json_data) {
	redisContext *rc = redisConnect("localhost", 6379);

	for(size_t i=0;i<cmds.size();i++) {
		redisAppendCommand(rc, cmds[i].c_str());
	}

	for(size_t i=0;i<cmds.size();i++) {
		redisReply *reply;
		redisGetReply(rc, (void **)&reply);

		if(reply->type == REDIS_REPLY_ERROR) {
			LOG_WARN("redis return error which msg:%s", reply->str);
			continue;
		}
		int pv_value = reply->type == REDIS_REPLY_NIL ? 0 : atoi(reply->str);

		Json::ArrayIndex index = (Json::ArrayIndex) i;
		json_data[index]["unit"] = times[i];
		json_data[index]["value"] = pv_value;
		LOG_DEBUG("start_time:%s, pv:%d", times[i].c_str(), pv_value);

		if(reply != NULL) {
			freeReplyObject(reply);
		}
	}

	redisFree(rc);
	return 0;
}

statistics_t *statistics_open(const char *path, const database_t *redis)
{
    /* allocate memory. */
    statistics_t *stats = (statistics_t *)malloc(sizeof(statistics_t));
    strcpy(stats->fname, path);

    /* open statistics file. */
    stats->fd = open(path, O_CREAT | O_RDWR | O_APPEND, S_IRUSR | S_IWUSR);
    if (stats->fd == -1) {
        log_error("open statistics file [%s] error, errno [%d]", path, errno);
        return NULL;
    }

    stats->conn = redisConnect(redis->ip, redis->port);
    if (stats->conn == NULL || stats->conn->err) {
        if (stats->conn != NULL) {
            log_error("Connection error: %s", stats->conn->errstr);
        } else {
            log_error("Connection error: can't allocate statistics redis context");
        }
        return NULL;
    }

    return stats;
}

void *Redis_create(char *host, int port) {
	redisContext *con;
	con = redisConnect(host, port);

	if (con == NULL || con->err) {
		if (con) {
			printf("Connection error: %s\n", con->errstr);
			redisFree(con);
		} else {
			printf("Connection error: can't allocate redis context\n");
		}
		exit(1);
	}

	Redis proto = { .host = host, .port = port, .con = con, .init = Redis_init,
			.readReply = Redis_readReply, .close = Redis_close, .command =
					Redis_command };

	Redis *m = calloc(1, sizeof(Redis));
	*m = proto;

	if (!m->init(m)) {
		// looks like it didn't initialize properly
		m->close(m);
		return NULL;
	} else {
		// all done, we made an object of any type
		return m;
	}

}

redisContext*
_nss_redis_redis_connect() {

  if (c)  {
    return c;
  }

  if(*_nss_redis_config.host == '/') {
    c = redisConnectUnix(_nss_redis_config.host);
  } else{
    c = redisConnect(_nss_redis_config.host, _nss_redis_config.port);
  }

  if(c->err) {
    redisFree(c);
    c = NULL;
    goto connected;
  }

  if(_nss_redis_redis_auth(c) == false) {
    c = NULL;
    goto connected;
  }

 connected:
  return c;
}

int emit_init()
{
    char *host, *buf = strdup(config.emit_option), *bp;
    int port;

    if ((bp = strchr(buf, '/')) == NULL) {
    	fprintf(stderr, "Redis emitter option is bad (no slash)\n");
	return (1);
    }
    *bp = 0;
    port = atoi(bp+1);
    host = buf;

    config.rediscon = redisConnect(host, port);
    if (config.rediscon->err) {
    	fprintf(stderr, "Cannot connect to Redis at %s:%d: %s\n",
		host, port, config.rediscon->errstr);
	return (1);
    }

    olog("[*] Connected to Redis at %s:%d with nsid=%s (topic=%s)\n\n",
    	host, port, config.nsid, config.emit_topic);

    free(buf);
    return (0);
}

/* Set up startup-time initialization */
static int post_config(apr_pool_t *p, apr_pool_t *plog, apr_pool_t *ptemp, server_rec *s)
{
    void *data;
    const char *userdata_key = "llzr_init";
    llzr_config *conf = ap_get_module_config(parms->server->module_config, &llzr_module);

    conf->redisconn = redisConnect(conf->redis_host, conf->redis_port);
    if (conf->redisconn == NULL || conf->redisconn->err) {
        if (conf->redisconn) {
            ap_log_error(APLOG_MARK, APLOG_WARNING, 0, NULL, "Redis error encountered %s", conf->redisconn->errstr);
            redisFree(c);
        } else {
            ap_log_error(APLOG_MARK, APLOG_WARNING, 0, NULL, "Unable to connect to redis server");
        }

        /* We don't want to kill our site if Redis goes down, but we would be susceptible to attack */
    }

    apr_pool_userdata_get(&data, userdata_key, s->process->pool);
    if (!data) {
	   apr_pool_userdata_set((const void *)1, userdata_key,apr_pool_cleanup_null, s->process->pool);
	       return OK;
    }

    /* Lets get details from the MPM */
    ap_log_error(APLOG_MARK, APLOG_NOTICE, 0, NULL, MODULE_NAME " " MODULE_VERSION " started");
    ap_mpm_query(AP_MPMQ_HARD_LIMIT_THREADS, &thread_limit);
    ap_mpm_query(AP_MPMQ_HARD_LIMIT_DAEMONS, &server_limit);

    return OK;
}

/*
 * connect to Redis server
 */
static redisContext * oidc_cache_redis_connect(request_rec *r,
		oidc_cache_cfg_redis_t *context) {

	/* see if we already have a connection by looking it up in the process context */
	redisContext *ctx = NULL;
	apr_pool_userdata_get((void **) &ctx, OIDC_CACHE_REDIS_CONTEXT,
			r->server->process->pool);

	if (ctx == NULL) {

		/* no connection, connect to the configured Redis server */
		ctx = redisConnect(context->host_str, context->port);

		/* check for errors */
		if ((ctx == NULL) || (ctx->err != 0)) {
			oidc_error(r, "failed to connect to Redis server (%s:%d): '%s'",
					context->host_str, context->port, ctx->errstr);
			return NULL;
		}

		/* store the connection in the process context */
		apr_pool_userdata_set(ctx, OIDC_CACHE_REDIS_CONTEXT,
				(apr_status_t (*)(void *)) redisFree, r->server->process->pool);

		/* log the connection */
		oidc_debug(r, "successfully connected to Redis server (%s:%d)",
				context->host_str, context->port);
	}

	return ctx;
}

int git_odb_backend_hiredis(git_odb_backend **backend_out, const char *host, int port)
{
    hiredis_backend *backend;

    backend = calloc(1, sizeof (hiredis_backend));
    if (backend == NULL)
        return GITERR_NOMEMORY;

    backend->db = redisConnect(host, port);
    if (backend->db->err) {
		free(backend);
		return GIT_ERROR;
	}

    backend->parent.read = &hiredis_backend__read;
    backend->parent.read_prefix = &hiredis_backend__read_prefix;
    backend->parent.read_header = &hiredis_backend__read_header;
    backend->parent.write = &hiredis_backend__write;
    backend->parent.exists = &hiredis_backend__exists;
    backend->parent.free = &hiredis_backend__free;

    *backend_out = (git_odb_backend *) backend;

    return GIT_OK;
}

void write_redis(const char * key, char *value, int len) {
	const char * ip = "127.0.0.1";
	int port = 6379;
	int dbno = 1;
	redisContext *con = redisConnect(ip, port);
	if (con->err) {
		redisFree(con);
		printf("Connection error: %s", con->errstr);
		exit(-1);
	}

	printf("Writing...");
	time_t start, end;
	start = clock();

	void *reply0 = redisCommand(con, "select %d", dbno);
	void *reply1 = redisCommand(con, "SET key:%s %b", key, value,
			len * sizeof(char));

	end = clock();
	printf("%f seconds.\n", (double) (end - start) / CLOCKS_PER_SEC);

	freeReplyObject(reply0);
	freeReplyObject(reply1);
	redisFree(con);
}

int redis()
{
	char * insert_command;
	redisContext baglam;
	baglam = redisConnect("127.0.0.1", 6379);
	if (baglam == NULL)
	{
		printf("Baglanamadi!");
	}
	else
	{
		printf("Baglandi!");

	}
/*	insert_command = malloc(sizeof(1024));
	sprintf(insert_command,
			"INSERT INTO \"NODE-software\" VALUES (%s, '%s', '%s', '%s', '%s', '%s', \
			'%s', '%s','%s', '%s', %d, \
			'%s', '%s','%s', '%s','%s', '%s', \
			'%s', '%s', '%s', '%s');" \
			,ms4db_ptr->resourceType ,ms4db_ptr->name, m2mSmGenerateRandomId(BIG_SIZE), (char *)ms4db_ptr->creationTime, (char *)ms4db_ptr->lastModifiedTime, "",\
			"", (char *)ms4db_ptr->expirationTime,"", "", ms4db_ptr->mgmtDefinition, \
			ms4db_ptr->name, "/opt/m2m/update/", "", ms4db_ptr->version, ms4db_ptr->name, ms4db_ptr->url, \
		    ms4db_ptr->install?"true":"false", ms4db_ptr->uninstall?"true":"false", ms4db_ptr->activate?"true":"false", ms4db_ptr->deactivate?"true":"false");

	resultPtr = execCommand(conPtr, insert_command);
	DEBUG("Command: %s", insert_command);

	if(resultPtr != NULL)
	{
		free(resultPtr);
	}*/
	return 1;
}

MHIREDIS_T MHiRedis::connect(const std::string& host, const int port,
                             const std::sting& password, const int db) {
    context_ = redisConnect(host_ip_.c_str(), port_);

    if (!context_) {
        cleanUp();
        return MH_CONNECT_RET_NULL;
    }

    if (context_->err != 0) {
        // log errstr

        cleanUp();
        return MH_CONNECT_ERROR;
    }
    if (!passwork.emply()) {
        std::string auth_comm = "AUTH " + password;
        MHIREDIS_T ret = execCommand(auth_comm, NULL);
        if (ret != MH_SUCCESS) {
            return MH_AUTH_FAILED;
        }
    }

    std::string sel_comm = "SELECT " + itoa(db);
    ret = execCommand(sel_comm, NULL);
    if (ret != MH_SUCCESS) {
        return MH_SEL_FAILED;
    }

    return MH_SUCCESS;
}