/* 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;
}

/* 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;
			dbgprintf("nsdsel_gtls: data already present in buffer, initiating "
				  "dummy select %p->iBufferRcvReady=%d\n",
				  pThis, 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;
}

/* set our parent, the tcpsrv object */
static rsRetVal
SetTcpsrv(tcps_sess_t *pThis, tcpsrv_t *pSrv)
{
	DEFiRet;
	ISOBJ_TYPE_assert(pThis, tcps_sess);
	ISOBJ_TYPE_assert(pSrv, tcpsrv);
	pThis->pSrv = pSrv;
	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;
			--pThis->iBufferRcvReady; /* one "pseudo-read" less */
			dbgprintf("nsdl_gtls: dummy read, decermenting %p->iBufRcvReady, now %d\n",
				   pThis, pThis->iBufferRcvReady);
			FINALIZE;
		}
		if(pNsdGTLS->rtryCall == gtlsRtry_handshake) {
			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;
		}
		else if(pNsdGTLS->rtryCall == gtlsRtry_recv) {
			iRet = doRetry(pNsdGTLS);
			if(iRet == RS_RET_OK) {
				*pbIsReady = 0;
				FINALIZE;
			}
		}

		/* now we must ensure that we do not fall back to PTCP if we have
		 * done a "dummy" select. In that case, we know when the predicate
		 * is not matched here, we do not have data available for this
		 * socket. -- rgerhards, 2010-11-20
		 */
		if(pThis->iBufferRcvReady) {
			dbgprintf("nsd_gtls: dummy read, buffer not available for this FD\n");
			*pbIsReady = 0;
			FINALIZE;
		}
	}

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

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;
}

/* Set this thread to "always running" state (can not be unset)
 * rgerhards, 2009-07-20
 */
rsRetVal
wtiSetAlwaysRunning(wti_t *pThis)
{
	ISOBJ_TYPE_assert(pThis, wti);
	pThis->bAlwaysRunning = RSTRUE;
	return RS_RET_OK;
}

/* Specify if we should do standard PRI parsing before we pass the data
 * down to the parser module.
 */
static rsRetVal
SetDoPRIParsing(parser_t *pThis, int bDoIt)
{
	ISOBJ_TYPE_assert(pThis, parser);
	pThis->bDoPRIParsing = bDoIt;
	return RS_RET_OK;
}

/* open the listen sockets */
static rsRetVal
doOpenLstnSocks(tcpsrv_t *pSrv)
{
	gsssrv_t *pGSrv;
	DEFiRet;

	ISOBJ_TYPE_assert(pSrv, tcpsrv);
	pGSrv = pSrv->pUsr;
	assert(pGSrv != NULL);

	/* first apply some config settings */
	if(pGSrv->allowedMethods) {
		if(pGSrv->allowedMethods & ALLOWEDMETHOD_GSS) {
			if(TCPSessGSSInit()) {
				errmsg.LogError(0, NO_ERRCODE, "GSS-API initialization failed\n");
				pGSrv->allowedMethods &= ~(ALLOWEDMETHOD_GSS);
			}
		}
		if(pGSrv->allowedMethods) {
			/* fallback to plain TCP */
			CHKiRet(tcpsrv.create_tcp_socket(pSrv));
		} else {
			ABORT_FINALIZE(RS_RET_GSS_ERR);
		}
	}

finalize_it:
	RETiRet;
}

/* ConstructionFinalizer */
static rsRetVal
tcpsrvConstructFinalize(tcpsrv_t *pThis)
{
	DEFiRet;
	ISOBJ_TYPE_assert(pThis, tcpsrv);

	/* prepare network stream subsystem */
	CHKiRet(netstrms.Construct(&pThis->pNS));
	if(pThis->pszDrvrName != NULL)
		CHKiRet(netstrms.SetDrvrName(pThis->pNS, pThis->pszDrvrName));
	CHKiRet(netstrms.SetDrvrMode(pThis->pNS, pThis->iDrvrMode));
	if(pThis->pszDrvrAuthMode != NULL)
		CHKiRet(netstrms.SetDrvrAuthMode(pThis->pNS, pThis->pszDrvrAuthMode));
	if(pThis->pPermPeers != NULL)
		CHKiRet(netstrms.SetDrvrPermPeers(pThis->pNS, pThis->pPermPeers));
	CHKiRet(netstrms.ConstructFinalize(pThis->pNS));

	/* set up listeners */
	CHKmalloc(pThis->ppLstn = calloc(pThis->iLstnMax, sizeof(netstrm_t*)));
	CHKmalloc(pThis->ppLstnPort = calloc(pThis->iLstnMax, sizeof(tcpLstnPortList_t*)));
	iRet = pThis->OpenLstnSocks(pThis);

finalize_it:
	if(iRet != RS_RET_OK) {
		if(pThis->pNS != NULL)
			netstrms.Destruct(&pThis->pNS);
		errmsg.LogError(0, iRet, "tcpsrv could not create listener (inputname: '%s')",
				(pThis->pszInputName == NULL) ? (uchar*)"*UNSET*" : pThis->pszInputName);
	}
	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;
}

wtiCancelThrd(wti_t *pThis, const uchar *const cancelobj)
{
	DEFiRet;

	ISOBJ_TYPE_assert(pThis, wti);

	if(wtiGetState(pThis)) {
		LogMsg(0, RS_RET_ERR, LOG_WARNING, "%s: need to do cooperative cancellation "
			"- some data may be lost, increase timeout?", cancelobj);
		/* we first try the cooperative "cancel" interface */
		pthread_kill(pThis->thrdID, SIGTTIN);
		DBGPRINTF("sent SIGTTIN to worker thread %p, giving it a chance to terminate\n",
			(void *) pThis->thrdID);
		srSleep(0, 10000);
	}

	if(wtiGetState(pThis)) {
		LogMsg(0, RS_RET_ERR, LOG_WARNING, "%s: need to do hard cancellation", cancelobj);
		DBGPRINTF("cooperative worker termination failed, using cancellation...\n");
		DBGOPRINT((obj_t*) pThis, "canceling worker thread\n");
		pthread_cancel(pThis->thrdID);
		/* now wait until the thread terminates... */
		while(wtiGetState(pThis)) {
			srSleep(0, 10000);
		}
	}

	RETiRet;
}

/* Initialize TCP listener socket for a single port
 * rgerhards, 2009-05-21
 */
static inline rsRetVal
initTCPListener(tcpsrv_t *pThis, tcpLstnPortList_t *pPortEntry)
{
	DEFiRet;
	uchar *TCPLstnPort;

	ISOBJ_TYPE_assert(pThis, tcpsrv);
	assert(pPortEntry != NULL);

	if(!ustrcmp(pPortEntry->pszPort, UCHAR_CONSTANT("0")))
		TCPLstnPort = UCHAR_CONSTANT("514");
		/* use default - we can not do service db update, because there is
		 * no IANA-assignment for syslog/tcp. In the long term, we might
		 * re-use RFC 3195 port of 601, but that would probably break to
		 * many existing configurations.
		 * rgerhards, 2007-06-28
		 */
	else
		TCPLstnPort = pPortEntry->pszPort;

	// pPortEntry->pszAddr = NULL ==> bind to all interfaces
        CHKiRet(netstrm.LstnInit(pThis->pNS, (void*)pPortEntry, addTcpLstn, TCPLstnPort, pPortEntry->pszAddr, pThis->iSessMax));

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;
}

/* configure TCP listener settings.
 * Note: pszPort is handed over to us - the caller MUST NOT free it!
 * rgerhards, 2008-03-20
 */
static rsRetVal
configureTCPListen(tcpsrv_t *pThis, uchar *pszPort, int bSuppOctetFram, uchar *pszAddr)
{
	int i;
	uchar *pPort = pszPort;
	DEFiRet;

	assert(pszPort != NULL);
	ISOBJ_TYPE_assert(pThis, tcpsrv);

	/* extract port */
	i = 0;
	while(isdigit((int) *pPort)) {
		i = i * 10 + *pPort++ - '0';
	}

	if(i >= 0 && i <= 65535) {
		CHKiRet(addNewLstnPort(pThis, pszPort, bSuppOctetFram, pszAddr));
	} else {
		errmsg.LogError(0, NO_ERRCODE, "Invalid TCP listen port %s - ignored.\n", pszPort);
	}

finalize_it:
	RETiRet;
}

/* Initialize TCP sockets (for listener) and listens on them */
static rsRetVal
create_tcp_socket(tcpsrv_t *pThis)
{
	DEFiRet;
	rsRetVal localRet;
	tcpLstnPortList_t *pEntry;

	ISOBJ_TYPE_assert(pThis, tcpsrv);

	/* init all configured ports */
	pEntry = pThis->pLstnPorts;
	while(pEntry != NULL) {
		localRet = initTCPListener(pThis, pEntry);
		if(localRet != RS_RET_OK) {
			errmsg.LogError(0, localRet, "Could not create tcp listener, ignoring port %s bind-address %s.", pEntry->pszPort, pEntry->pszAddr);
		}
		pEntry = pEntry->pNext;
	}

	/* OK, we had success. Now it is also time to
	 * initialize our connections
	 */
	if(TCPSessTblInit(pThis) != 0) {
		/* OK, we are in some trouble - we could not initialize the
		 * session table, so we can not continue. We need to free all
		 * we have assigned so far, because we can not really use it...
		 */
		errmsg.LogError(0, RS_RET_ERR, "Could not initialize TCP session table, suspending TCP message reception.");
		ABORT_FINALIZE(RS_RET_ERR);
	}

finalize_it:
	RETiRet;
}

static rsRetVal
DataRcvd(tcps_sess_t *pThis, char *pData, size_t iLen)
{
	multi_submit_t multiSub;
	msg_t *pMsgs[NUM_MULTISUB];
	struct syslogTime stTime;
	time_t ttGenTime;
	char *pEnd;
	DEFiRet;

	ISOBJ_TYPE_assert(pThis, tcps_sess);
	assert(pData != NULL);
	assert(iLen > 0);

	datetime.getCurrTime(&stTime, &ttGenTime, TIME_IN_LOCALTIME);
	multiSub.ppMsgs = pMsgs;
	multiSub.maxElem = NUM_MULTISUB;
	multiSub.nElem = 0;

	 /* We now copy the message to the session buffer. */
	pEnd = pData + iLen; /* this is one off, which is intensional */

	while(pData < pEnd) {
		CHKiRet(processDataRcvd(pThis, *pData++, &stTime, ttGenTime, &multiSub));
	}
	iRet = multiSubmitFlush(&multiSub);

finalize_it:
	RETiRet;
}

/* add new listener port to listener port list
 * rgerhards, 2009-05-21
 */
static inline rsRetVal
addNewLstnPort(tcpsrv_t *pThis, uchar *pszPort, int bSuppOctetFram, uchar *pszAddr)
{
	tcpLstnPortList_t *pEntry;
	uchar statname[64];
	DEFiRet;

	ISOBJ_TYPE_assert(pThis, tcpsrv);

	/* create entry */
	CHKmalloc(pEntry = MALLOC(sizeof(tcpLstnPortList_t)));
	if((pEntry->pszPort = ustrdup(pszPort)) == NULL) {
		DBGPRINTF("tcpsrv/addNewLstnPort: OOM in strdup()\n");
		free(pEntry);
		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
	}

        pEntry->pszAddr = NULL;
        /* only if a bind adress is defined copy it in struct */
        if (pszAddr != NULL) {
		if((pEntry->pszAddr = ustrdup(pszAddr)) == NULL) {
			DBGPRINTF("tcpsrv/addNewLstnPort: OOM in strdup() 2\n");
			free(pEntry->pszPort);
			free(pEntry);
			ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
		}
	}

	strcpy((char*)pEntry->dfltTZ, (char*)pThis->dfltTZ);
	pEntry->bSPFramingFix = pThis->bSPFramingFix;
	pEntry->pSrv = pThis;
	pEntry->pRuleset = pThis->pRuleset;
	pEntry->bSuppOctetFram = bSuppOctetFram;

	/* we need to create a property */ 
	CHKiRet(prop.Construct(&pEntry->pInputName));
	CHKiRet(prop.SetString(pEntry->pInputName, pThis->pszInputName, ustrlen(pThis->pszInputName)));
	CHKiRet(prop.ConstructFinalize(pEntry->pInputName));

	/* and add to list */
	pEntry->pNext = pThis->pLstnPorts;
	pThis->pLstnPorts = pEntry;

	/* support statistics gathering */
	CHKiRet(statsobj.Construct(&(pEntry->stats)));
	snprintf((char*)statname, sizeof(statname), "%s(%s)", pThis->pszInputName, pszPort);
	statname[sizeof(statname)-1] = '\0'; /* just to be on the save side... */
	CHKiRet(statsobj.SetName(pEntry->stats, statname));
	CHKiRet(statsobj.SetOrigin(pEntry->stats, pThis->pszOrigin));
	CHKiRet(ratelimitNew(&pEntry->ratelimiter, "tcperver", NULL));
	ratelimitSetLinuxLike(pEntry->ratelimiter, pThis->ratelimitInterval, pThis->ratelimitBurst);
	ratelimitSetThreadSafe(pEntry->ratelimiter);
	STATSCOUNTER_INIT(pEntry->ctrSubmit, pEntry->mutCtrSubmit);
	CHKiRet(statsobj.AddCounter(pEntry->stats, UCHAR_CONSTANT("submitted"),
		ctrType_IntCtr, CTR_FLAG_RESETTABLE, &(pEntry->ctrSubmit)));
	CHKiRet(statsobj.ConstructFinalize(pEntry->stats));

finalize_it:
	RETiRet;
}

/* open the listen sockets */
static int*
doOpenLstnSocks(tcpsrv_t *pSrv)
{
	int *pRet = NULL;
	gsssrv_t *pGSrv;

	ISOBJ_TYPE_assert(pSrv, tcpsrv);
	pGSrv = pSrv->pUsr;
	assert(pGSrv != NULL);

	/* first apply some config settings */
	if(pGSrv->allowedMethods) {
		if(pGSrv->allowedMethods & ALLOWEDMETHOD_GSS) {
			if(TCPSessGSSInit()) {
				errmsg.LogError(NO_ERRCODE, "GSS-API initialization failed\n");
				pGSrv->allowedMethods &= ~(ALLOWEDMETHOD_GSS);
			}
		}
		if(pGSrv->allowedMethods) {
			/* fallback to plain TCP */
			if((pRet =  tcpsrv.create_tcp_socket(pSrv)) != NULL) {
				dbgprintf("Opened %d syslog TCP port(s).\n", *pRet);
			}
		}
	}

	return pRet;
}

/* Cancel the thread. If the thread is not running. But it is save and legal to
 * call wtiCancelThrd() in such situations. This function only returns when the
 * thread has terminated. Else we may get race conditions all over the code...
 * Note that when waiting for the thread to terminate, we do a busy wait, checking
 * progress every 10ms. It is very unlikely that we will ever cancel a thread
 * and, if so, it will only happen at the end of the rsyslog run. So doing this
 * kind of non-optimal wait is considered preferable over using condition variables.
 * rgerhards, 2008-02-26
 */
rsRetVal
wtiCancelThrd(wti_t *pThis)
{
	DEFiRet;

	ISOBJ_TYPE_assert(pThis, wti);


	if(wtiGetState(pThis)) {
		/* we first try the cooperative "cancel" interface */
		pthread_kill(pThis->thrdID, SIGTTIN);
		DBGPRINTF("sent SIGTTIN to worker thread 0x%x, giving it a chance to terminate\n", (unsigned) pThis->thrdID);
		srSleep(0, 10000);
	}

	if(wtiGetState(pThis)) {
		DBGPRINTF("cooperative worker termination failed, using cancellation...\n");
		DBGOPRINT((obj_t*) pThis, "canceling worker thread\n");
		pthread_cancel(pThis->thrdID);
		/* now wait until the thread terminates... */
		while(wtiGetState(pThis)) {
			srSleep(0, 10000);
		}
	}

	RETiRet;
}

/* Specify if we should do standard message sanitazion before we pass the data
 * down to the parser.
 */
static rsRetVal
SetDoSanitazion(parser_t *pThis, int bDoIt)
{
	ISOBJ_TYPE_assert(pThis, parser);
	pThis->bDoSanitazion = bDoIt;
	return RS_RET_OK;
}