/* Start a new thread and add it to the list of currently
 * executing threads. It is added at the end of the list.
 * rgerhards, 2007-12-14
 */
rsRetVal thrdCreate(rsRetVal (*thrdMain)(thrdInfo_t*), rsRetVal(*afterRun)(thrdInfo_t *), sbool bNeedsCancel, uchar *name)
{
	DEFiRet;
	thrdInfo_t *pThis;

	assert(thrdMain != NULL);

	CHKiRet(thrdConstruct(&pThis));
	pThis->bIsActive = 1;
	pThis->pUsrThrdMain = thrdMain;
	pThis->pAfterRun = afterRun;
	pThis->bNeedsCancel = bNeedsCancel;
	pThis->name = ustrdup(name);
	pthread_create(&pThis->thrdID,
#ifdef HAVE_PTHREAD_SETSCHEDPARAM
			   &default_thread_attr,
#else
			   NULL,
#endif
			   thrdStarter, pThis);
	CHKiRet(llAppend(&llThrds, NULL, pThis));

finalize_it:
	RETiRet;
}

/* enqueue the the kernel message into the message queue.
 * The provided msg string is not freed - thus must be done
 * by the caller.
 * rgerhards, 2008-04-12
 */
static rsRetVal
enqMsg(uchar *msg, uchar* pszTag, syslog_pri_t pri, struct timeval *tp, struct json_object *json)
{
	struct syslogTime st;
	msg_t *pMsg;
	DEFiRet;

	assert(msg != NULL);
	assert(pszTag != NULL);

	if(tp == NULL) {
		CHKiRet(msgConstruct(&pMsg));
	} else {
		datetime.timeval2syslogTime(tp, &st);
		CHKiRet(msgConstructWithTime(&pMsg, &st, tp->tv_sec));
	}
	MsgSetFlowControlType(pMsg, eFLOWCTL_LIGHT_DELAY);
	MsgSetInputName(pMsg, pInputName);
	MsgSetRawMsgWOSize(pMsg, (char*)msg);
	MsgSetMSGoffs(pMsg, 0);	/* we do not have a header... */
	MsgSetRcvFrom(pMsg, glbl.GetLocalHostNameProp());
	MsgSetRcvFromIP(pMsg, pLocalHostIP);
	MsgSetHOSTNAME(pMsg, glbl.GetLocalHostName(), ustrlen(glbl.GetLocalHostName()));
	MsgSetTAG(pMsg, pszTag, ustrlen(pszTag));
	msgSetPRI(pMsg, pri);
	pMsg->json = json;
	CHKiRet(submitMsg(pMsg));

finalize_it:
	RETiRet;
}

/* Parse a number from the configuration line.
 * rgerhards, 2007-07-31
 */
static rsRetVal doGetInt(uchar **pp, rsRetVal (*pSetHdlr)(void*, uid_t), void *pVal)
{
	uchar *p;
	DEFiRet;
	int64 i;	

	assert(pp != NULL);
	assert(*pp != NULL);
	
	CHKiRet(parseIntVal(pp, &i));
	p = *pp;

	if(pSetHdlr == NULL) {
		/* we should set value directly to var */
		*((int*)pVal) = (int) i;
	} else {
		/* we set value via a set function */
		CHKiRet(pSetHdlr(pVal, (int) i));
	}

	*pp = p;

finalize_it:
	RETiRet;
}

/* Construction finalizer
 * rgerhards, 2008-01-17
 */
rsRetVal
wtpConstructFinalize(wtp_t *pThis)
{
	DEFiRet;
	int i;
	uchar pszBuf[64];
	size_t lenBuf;
	wti_t *pWti;

	ISOBJ_TYPE_assert(pThis, wtp);

	dbgprintf("%s: finalizing construction of worker thread pool\n", wtpGetDbgHdr(pThis));
	/* alloc and construct workers - this can only be done in finalizer as we previously do
	 * not know the max number of workers
	 */
	if((pThis->pWrkr = malloc(sizeof(wti_t*) * pThis->iNumWorkerThreads)) == NULL)
		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);

	for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) {
		CHKiRet(wtiConstruct(&pThis->pWrkr[i]));
		pWti = pThis->pWrkr[i];
		lenBuf = snprintf((char*)pszBuf, sizeof(pszBuf), "%s/w%d", wtpGetDbgHdr(pThis), i);
		CHKiRet(wtiSetDbgHdr(pWti, pszBuf, lenBuf));
		CHKiRet(wtiSetpWtp(pWti, pThis));
		CHKiRet(wtiConstructFinalize(pWti));
	}
		

finalize_it:
	RETiRet;
}

/* Parse a number from the configuration line.
 * rgerhards, 2007-07-31
 */
static rsRetVal doGetInt(uchar **pp, rsRetVal (*pSetHdlr)(void*, uid_t), void *pVal)
{
	uchar *p;
	DEFiRet;
	int64 i;	
	uchar errMsg[256];	/* for dynamic error messages */

	assert(pp != NULL);
	assert(*pp != NULL);
	
	CHKiRet(doGetSize(pp, NULL,&i));
	p = *pp;
	if(i > 2147483648ll) { /*2^31*/
		snprintf((char*) errMsg, sizeof(errMsg)/sizeof(uchar),
		         "value %lld too large for integer argument.", i);
		errmsg.LogError(0, RS_RET_INVALID_VALUE, "%s", errMsg);
		ABORT_FINALIZE(RS_RET_INVALID_VALUE);
	}

	if(pSetHdlr == NULL) {
		/* we should set value directly to var */
		*((int*)pVal) = (int) i;
	} else {
		/* we set value via a set function */
		CHKiRet(pSetHdlr(pVal, (int) i));
	}

	*pp = p;

finalize_it:
	RETiRet;
}

/* Add a socket to the select set */
static rsRetVal
Add(nsdsel_t *pNsdsel, nsd_t *pNsd, nsdsel_waitOp_t waitOp)
{
	DEFiRet;
	nsdsel_gtls_t *pThis = (nsdsel_gtls_t*) pNsdsel;
	nsd_gtls_t *pNsdGTLS = (nsd_gtls_t*) pNsd;

	ISOBJ_TYPE_assert(pThis, nsdsel_gtls);
	ISOBJ_TYPE_assert(pNsdGTLS, nsd_gtls);
	if(pNsdGTLS->iMode == 1) {
		if(waitOp == NSDSEL_RD && gtlsHasRcvInBuffer(pNsdGTLS)) {
			++pThis->iBufferRcvReady;
			FINALIZE;
		}
		if(pNsdGTLS->rtryCall != gtlsRtry_None) {
			if(gnutls_record_get_direction(pNsdGTLS->sess) == 0) {
				CHKiRet(nsdsel_ptcp.Add(pThis->pTcp, pNsdGTLS->pTcp, NSDSEL_RD));
			} else {
				CHKiRet(nsdsel_ptcp.Add(pThis->pTcp, pNsdGTLS->pTcp, NSDSEL_WR));
			}
			FINALIZE;
		}
	}

	/* if we reach this point, we need no special handling */
	CHKiRet(nsdsel_ptcp.Add(pThis->pTcp, pNsdGTLS->pTcp, waitOp));

finalize_it:
	RETiRet;
}

/* parse the audit record and create libee structure
 */
static rsRetVal
audit_parse(uchar *buf, struct json_object **jsonRoot)
{
	struct json_object *json;
	struct json_object *jval;
	char name[1024];
	char val[1024];
	DEFiRet;

	*jsonRoot = json_object_new_object();
	if(*jsonRoot == NULL) {
		ABORT_FINALIZE(RS_RET_ERR);
	}
	json = json_object_new_object();
	json_object_object_add(*jsonRoot, "data", json);

	while(*buf) {
		CHKiRet(parseName(&buf, name, sizeof(name)));
		if(*buf != '=') {
			ABORT_FINALIZE(RS_RET_ERR);
		}
		++buf;
		CHKiRet(parseValue(&buf, val, sizeof(val)));
		jval = json_object_new_string(val);
		json_object_object_add(json, name, jval);
	}
	

finalize_it:
	RETiRet;
}

/* Parse and a word config line option. A word is a consequtive
 * sequence of non-whitespace characters. pVal must be
 * a pointer to a string which is to receive the option
 * value. The returned string must be freed by the caller.
 * rgerhards, 2007-09-07
 * To facilitate multiple instances of the same command line
 * directive, doGetWord() now checks if pVal is already a
 * non-NULL pointer. If so, we assume it was created by a previous
 * incarnation and is automatically freed. This happens only when
 * no custom handler is defined. If it is, the customer handler
 * must do the cleanup. I have checked and this was al also memory
 * leak with some code. Obviously, not a large one. -- rgerhards, 2007-12-20
 * Just to clarify: if pVal is parsed to a custom handler, this handler
 * is responsible for freeing pVal. -- rgerhards, 2008-03-20
 */
static rsRetVal doGetWord(uchar **pp, rsRetVal (*pSetHdlr)(void*, uchar*), void *pVal)
{
	DEFiRet;
	cstr_t *pStrB = NULL;
	uchar *pNewVal;

	ASSERT(pp != NULL);
	ASSERT(*pp != NULL);

	CHKiRet(getWord(pp, &pStrB));
	CHKiRet(cstrConvSzStrAndDestruct(&pStrB, &pNewVal, 0));

	DBGPRINTF("doGetWord: get newval '%s' (len %d), hdlr %p\n",
		  pNewVal, (int) ustrlen(pNewVal), pSetHdlr);
	/* we got the word, now set it */
	if(pSetHdlr == NULL) {
		/* we should set value directly to var */
		if(*((uchar**)pVal) != NULL)
			free(*((uchar**)pVal)); /* free previous entry */
		*((uchar**)pVal) = pNewVal; /* set new one */
	} else {
		/* we set value via a set function */
		CHKiRet(pSetHdlr(pVal, pNewVal));
	}

	skipWhiteSpace(pp); /* skip over any whitespace */

finalize_it:
	if(iRet != RS_RET_OK) {
		if(pStrB != NULL)
			cstrDestruct(&pStrB);
	}

	RETiRet;
}

/* check if a socket is ready for IO */
static rsRetVal
IsReady(nsdsel_t *pNsdsel, nsd_t *pNsd, nsdsel_waitOp_t waitOp, int *pbIsReady)
{
	DEFiRet;
	nsdsel_gtls_t *pThis = (nsdsel_gtls_t*) pNsdsel;
	nsd_gtls_t *pNsdGTLS = (nsd_gtls_t*) pNsd;

	ISOBJ_TYPE_assert(pThis, nsdsel_gtls);
	ISOBJ_TYPE_assert(pNsdGTLS, nsd_gtls);
	if(pNsdGTLS->iMode == 1) {
		if(waitOp == NSDSEL_RD && gtlsHasRcvInBuffer(pNsdGTLS)) {
			*pbIsReady = 1;
			FINALIZE;
		}
		if(pNsdGTLS->rtryCall != gtlsRtry_None) {
			CHKiRet(doRetry(pNsdGTLS));
			/* we used this up for our own internal processing, so the socket
			 * is not ready from the upper layer point of view.
			 */
			*pbIsReady = 0;
			FINALIZE;
		}
	}

	CHKiRet(nsdsel_ptcp.IsReady(pThis->pTcp, pNsdGTLS->pTcp, waitOp, pbIsReady));

finalize_it:
	RETiRet;
}

/* this function can be used to obtain all stats lines. In this case,
 * a callback must be provided. This module than iterates over all objects and
 * submits each stats line to the callback. The callback has two parameters:
 * the first one is a caller-provided void*, the second one the cstr_t with the
 * line. If the callback reports an error, processing is stopped.
 */
static rsRetVal
getAllStatsLines(rsRetVal(*cb)(void*, cstr_t*), void *usrptr, statsFmtType_t fmt, int8_t bResetCtrs)
{
	statsobj_t *o;
	cstr_t *cstr;
	DEFiRet;

	for(o = objRoot ; o != NULL ; o = o->next) {
		switch(fmt) {
		case statsFmt_Legacy:
			CHKiRet(getStatsLine(o, &cstr, bResetCtrs));
			break;
		case statsFmt_CEE:
			CHKiRet(getStatsLineCEE(o, &cstr, 1, bResetCtrs));
			break;
		case statsFmt_JSON:
			CHKiRet(getStatsLineCEE(o, &cstr, 0, bResetCtrs));
			break;
		}
		CHKiRet(cb(usrptr, cstr));
		rsCStrDestruct(&cstr);
	}

finalize_it:
	RETiRet;
}

/* Construction finalizer
 * rgerhards, 2008-01-17
 */
rsRetVal
wtiConstructFinalize(wti_t *pThis)
{
	DEFiRet;
	int iDeqBatchSize;

	ISOBJ_TYPE_assert(pThis, wti);

	DBGPRINTF("%s: finalizing construction of worker instance data (for %d actions)\n",
		  wtiGetDbgHdr(pThis), iActionNbr);

	/* initialize our thread instance descriptor (no concurrency here) */
	pThis->bIsRunning = WRKTHRD_STOPPED;

	/* must use calloc as we need zero-init */
	CHKmalloc(pThis->actWrkrInfo = calloc(iActionNbr, sizeof(actWrkrInfo_t)));

	if(pThis->pWtp == NULL) {
		dbgprintf("wtiConstructFinalize: pWtp not set, this may be intentional\n");
		FINALIZE;
	}

	/* we now alloc the array for user pointers. We obtain the max from the queue itself. */
	CHKiRet(pThis->pWtp->pfGetDeqBatchSize(pThis->pWtp->pUsr, &iDeqBatchSize));
	CHKiRet(batchInit(&pThis->batch, iDeqBatchSize));

finalize_it:
	RETiRet;
}

/* enqueue the the kernel message into the message queue.
 * The provided msg string is not freed - thus must be done
 * by the caller.
 * rgerhards, 2008-04-12
 */
static rsRetVal
enqMsg(uchar *msg, uchar* pszTag, int iFacility, int iSeverity)
{
	DEFiRet;
	msg_t *pMsg;

	assert(msg != NULL);
	assert(pszTag != NULL);

	CHKiRet(msgConstruct(&pMsg));
	MsgSetFlowControlType(pMsg, eFLOWCTL_LIGHT_DELAY);
	MsgSetInputName(pMsg, pInputName);
	MsgSetRawMsgWOSize(pMsg, (char*)msg);
	MsgSetMSGoffs(pMsg, 0);	/* we do not have a header... */
	MsgSetRcvFrom(pMsg, glbl.GetLocalHostNameProp());
	MsgSetRcvFromIP(pMsg, pLocalHostIP);
	MsgSetHOSTNAME(pMsg, glbl.GetLocalHostName(), ustrlen(glbl.GetLocalHostName()));
	MsgSetTAG(pMsg, pszTag, ustrlen(pszTag));
	pMsg->iFacility = LOG_FAC(iFacility);
	pMsg->iSeverity = LOG_PRI(iSeverity);
	pMsg->bParseHOSTNAME = 0;
	CHKiRet(submitMsg(pMsg));

finalize_it:
	RETiRet;
}

/* get all the object's countes together with object name as one line.
 */
static rsRetVal
getStatsLine(statsobj_t *pThis, cstr_t **ppcstr, int8_t bResetCtrs)
{
	cstr_t *pcstr;
	ctr_t *pCtr;
	DEFiRet;

	CHKiRet(cstrConstruct(&pcstr));
	rsCStrAppendStr(pcstr, pThis->name);
	rsCStrAppendStrWithLen(pcstr, UCHAR_CONSTANT(": "), 2);

	/* now add all counters to this line */
	pthread_mutex_lock(&pThis->mutCtr);
	for(pCtr = pThis->ctrRoot ; pCtr != NULL ; pCtr = pCtr->next) {
		rsCStrAppendStr(pcstr, pCtr->name);
		cstrAppendChar(pcstr, '=');
		switch(pCtr->ctrType) {
		case ctrType_IntCtr:
			rsCStrAppendInt(pcstr, *(pCtr->val.pIntCtr)); // TODO: OK?????
			break;
		case ctrType_Int:
			rsCStrAppendInt(pcstr, *(pCtr->val.pInt));
			break;
		}
		cstrAppendChar(pcstr, ' ');
		resetResettableCtr(pCtr, bResetCtrs);
	}
	pthread_mutex_unlock(&pThis->mutCtr);

	CHKiRet(cstrFinalize(pcstr));
	*ppcstr = pcstr;

finalize_it:
	RETiRet;
}

/* accept an incoming connection request
 * The netstrm instance that had the incoming request must be provided. If
 * the connection request succeeds, a new netstrm object is created and 
 * passed back to the caller. The caller is responsible for destructing it.
 * pReq is the nsd_t obj that has the accept request.
 * rgerhards, 2008-04-21
 */
static rsRetVal
AcceptConnReq(netstrm_t *pThis, netstrm_t **ppNew)
{
	nsd_t *pNewNsd = NULL;
	DEFiRet;

	ISOBJ_TYPE_assert(pThis, netstrm);
	assert(ppNew != NULL);

	/* accept the new connection */
	CHKiRet(pThis->Drvr.AcceptConnReq(pThis->pDrvrData, &pNewNsd));
	/* construct our object so that we can use it... */
	CHKiRet(objUse(netstrms, DONT_LOAD_LIB)); /* use netstrms obj if not already done so */
	CHKiRet(netstrms.CreateStrm(pThis->pNS, ppNew));
	(*ppNew)->pDrvrData = pNewNsd;

finalize_it:
	if(iRet != RS_RET_OK) {
		/* the close may be redundant, but that doesn't hurt... */
		if(pNewNsd != NULL)
			pThis->Drvr.Destruct(&pNewNsd);
	}

	RETiRet;
}

/* helper to ochAddLine. Parses a comma-delimited field
 * The field is delimited by SP or comma. Leading whitespace
 * is "eaten" and does not become part of the field content.
 */
static rsRetVal get_Field(uchar **pp, uchar **pField)
{
	DEFiRet;
	register uchar *p;
	cstr_t *pStrB = NULL;

	assert(pp != NULL);
	assert(*pp != NULL);
	assert(pField != NULL);

	skip_Comma((char**)pp);
	p = *pp;

	CHKiRet(cstrConstruct(&pStrB));

	/* copy the field */
	while(*p && *p != ' ' && *p != ',') {
		CHKiRet(cstrAppendChar(pStrB, *p++));
	}

	*pp = p;
	CHKiRet(cstrFinalize(pStrB));
	CHKiRet(cstrConvSzStrAndDestruct(&pStrB, pField, 0));

finalize_it:
	if(iRet != RS_RET_OK) {
		if(pStrB != NULL)
			cstrDestruct(&pStrB);
	}

	RETiRet;
}

/* helper to ochAddLine. Parses everything from the
 * current position to the end of line and returns it
 * to the caller. Leading white space is removed, but
 * not trailing.
 */
static inline rsRetVal get_restOfLine(uchar **pp, uchar **pBuf)
{
	DEFiRet;
	register uchar *p;
	cstr_t *pStrB = NULL;

	assert(pp != NULL);
	assert(*pp != NULL);
	assert(pBuf != NULL);

	skip_Comma((char**)pp);
	p = *pp;

	CHKiRet(cstrConstruct(&pStrB));

	/* copy the field */
	while(*p) {
		CHKiRet(cstrAppendChar(pStrB, *p++));
	}

	*pp = p;
	CHKiRet(cstrFinalize(pStrB));
	CHKiRet(cstrConvSzStrAndDestruct(&pStrB, pBuf, 0));

finalize_it:
	if(iRet != RS_RET_OK) {
		if(pStrB != NULL)
			cstrDestruct(&pStrB);
	}

	RETiRet;
}

/* parse a whitespace-delimited word from the provided string. This is a
 * helper function for a number of syntaxes. The parsed value is returned
 * in ppStrB (which must be provided by caller).
 * rgerhards, 2008-02-14
 */
static rsRetVal
getWord(uchar **pp, cstr_t **ppStrB)
{
	DEFiRet;
	uchar *p;

	ASSERT(pp != NULL);
	ASSERT(*pp != NULL);
	ASSERT(ppStrB != NULL);

	CHKiRet(cstrConstruct(ppStrB));

	skipWhiteSpace(pp); /* skip over any whitespace */

	/* parse out the word */
	p = *pp;

	while(*p && !isspace((int) *p)) {
		CHKiRet(cstrAppendChar(*ppStrB, *p++));
	}
	CHKiRet(cstrFinalize(*ppStrB));

	*pp = p;

finalize_it:
	RETiRet;
}

/* Parse a number from the configuration line.
 * rgerhards, 2007-07-31
 */
static rsRetVal doGetInt(uchar **pp, rsRetVal (*pSetHdlr)(void*, uid_t), void *pVal)
{
	uchar *p;
	DEFiRet;
	int64 i;	

	assert(pp != NULL);
	assert(*pp != NULL);
	
	CHKiRet(doGetSize(pp, NULL,&i));
	p = *pp;
	if(i > 2147483648ll) { /*2^31*/
		LogError(0, RS_RET_INVALID_VALUE, 
		         "value %lld too large for integer argument.", i);
		ABORT_FINALIZE(RS_RET_INVALID_VALUE);
	}

	if(pSetHdlr == NULL) {
		/* we should set value directly to var */
		*((int*)pVal) = (int) i;
	} else {
		/* we set value via a set function */
		CHKiRet(pSetHdlr(pVal, (int) i));
	}

	*pp = p;

finalize_it:
	RETiRet;
}

/* parse a syslog name from the string. This is the generic code that is
 * called by the facility/severity functions. Note that we do not check the
 * validity of numerical values, something that should probably change over
 * time (TODO). -- rgerhards, 2008-02-14
 */
static rsRetVal
doSyslogName(uchar **pp, rsRetVal (*pSetHdlr)(void*, int),
	  	    void *pVal, syslogName_t *pNameTable)
{
	DEFiRet;
	cstr_t *pStrB;
	int iNewVal;

	ASSERT(pp != NULL);
	ASSERT(*pp != NULL);

	CHKiRet(getWord(pp, &pStrB)); /* get word */
	iNewVal = decodeSyslogName(cstrGetSzStr(pStrB), pNameTable);

	if(pSetHdlr == NULL) {
		/* we should set value directly to var */
		*((int*)pVal) = iNewVal; /* set new one */
	} else {
		/* we set value via a set function */
		CHKiRet(pSetHdlr(pVal, iNewVal));
	}

	skipWhiteSpace(pp); /* skip over any whitespace */

finalize_it:
	if(pStrB != NULL)
		rsCStrDestruct(&pStrB);

	RETiRet;
}

/* The following function writes the current log entry
 * to an established MySQL session.
 * Initially added 2004-10-28 mmeckelein
 */
rsRetVal writeMySQL(wrkrInstanceData_t *pWrkrData, uchar *psz)
{
	DEFiRet;

	/* see if we are ready to proceed */
	if(pWrkrData->hmysql == NULL) {
		CHKiRet(initMySQL(pWrkrData, 0));
		
	}

	/* try insert */
	if(mysql_query(pWrkrData->hmysql, (char*)psz)) {
		/* error occured, try to re-init connection and retry */
		closeMySQL(pWrkrData); /* close the current handle */
		CHKiRet(initMySQL(pWrkrData, 0)); /* try to re-open */
		if(mysql_query(pWrkrData->hmysql, (char*)psz)) { /* re-try insert */
			/* we failed, giving up for now */
			reportDBError(pWrkrData, 0);
			closeMySQL(pWrkrData); /* free ressources */
			ABORT_FINALIZE(RS_RET_SUSPENDED);
		}
	}

finalize_it:
	if(iRet == RS_RET_OK) {
		pWrkrData->uLastMySQLErrno = 0; /* reset error for error supression */
	}

	RETiRet;
}