/********************************************************************/
/*                                                                  */
/*   MQTEST is a derivative of MQPut2 desinged for testing rather   */
/*   than as a performance measurement driver.  It reads each file  */
/*   in order and writes it to the specified queue.  Each file is   */
/*   written once. The parameter file is the same as used by the    */
/*   MQPut2 program.  Certain parameters such as sleep interval are */
/*   ignored.                                                       */
/*                                                                  */
/*   MQTEST has 1 required input parameter                          */
/*                                                                  */
/*      -f name of the parameter input file                         */
/*                                                                  */
/*   The input parameter file contains all values, including the    */
/*   name of the queue and queue manager to write data to, the      */
/*   total number of messages to write, any MQMD parameters and     */
/*   a list of files which contain the message data.                */
/*                                                                  */
/********************************************************************/

#include <stdio.h>
#include <stdlib.h>
#include <ctype.h>
#include <string.h>

#ifdef WIN32
#include <windows.h>
#endif

#ifdef SOLARIS
#include <ctype.h>
#include <unistd.h>
#endif

/* includes for MQI */
#include <cmqc.h>

/* common subroutines include */
#include "int64defs.h"
#include "comsubs.h"

/* parameter file processing routines */
#include "putparms.h"
#include "parmline.h"

/* MQ subroutines include */
#include "qsubs.h"
#include "rfhsubs.h"

#ifndef _WIN32
void Sleep(int amount)
{
	usleep(amount*1000);
}
#endif

static char copyright[] = "(C) Copyright IBM Corp, 2001 - 2006\n";
static char Version[]=\
"@(#)MQTest V1.0.1 - Message Broker Test driver tool  - Jim MacNair ";

#ifdef _DEBUG
static char Level[]="mqtest.c V1.0.1 Debug version ("__DATE__" "__TIME__")\n";
#else
static char Level[]="mqtest.c V1.0.1 Release version ("__DATE__" "__TIME__")\n";
#endif

	MQHCONN			qm=0;			/* queue manager connection handle */
	MQHOBJ			q=0;			/* queue handle used for mqputs    */
	static int		filelist = 0;	/* found [filelist] separator      */

	/* variables to keep track of parameter file sections */
	int		foundfiles=0;
	int		foundUsr=0;

	/* global error switch */
	int			err=0;

	/* counters and statistics */
	MQCHAR8		puttime;

	/* state information */
	int		groupOpen = 0;
	int		uow=0;

	/* name of the connected qm and open queue */
	char	connectName[MQ_Q_MGR_NAME_LENGTH + 1];
	char	openName[MQ_Q_NAME_LENGTH + 1];
	char	openRemoteName[MQ_Q_NAME_LENGTH + 1];

char * searchForDelimiter(char * msgPtr, const int maxlen, putParms * parms)

{
	int		i=0;
	char *	newPtr=NULL;

	/* return if there is no delimiter specified */
	if (0 == parms->delimiterLen)
	{
		return NULL;
	}

	while (1)
	{
		while ((i < (maxlen - parms->delimiterLen)) && (msgPtr[i] != parms->delimiter[0]))
		{
			i++;
		}

		/* did we run out of message to check? */
		if (i < (maxlen - parms->delimiterLen))
		{
			/* we found a delimiter that we need to check */
			/* check if we have more delimiter characters to compare */
			if (1 == parms->delimiterLen)
			{
				/* no, found a delimiter */
				/* set the new address pointer */
				newPtr = msgPtr + i;
				break;
			}
			else
			{
				/* do a full memory compare */
				if (0 == memcmp(msgPtr + i, parms->delimiter, parms->delimiterLen))
				{
					/* we have found a delimiter */
					newPtr = msgPtr + i;
					break;
				}
			}
		}
		else
		{
			break;
		}

		/* go on to the next entry */
		i++;
	}

	return newPtr;
}

void discQM(putParms * parms)

{
	MQLONG	compcode=0;
	MQLONG	reason=0;

	if (0 == qm)
	{
		return;
	}

	/* Disconnect from the queue manager */
	printf("disconnecting from the queue manager %s\n", parms->qmname);
	MQDISC(&qm, &compcode, &reason);

	/* check for errors */
	checkerror("MQDISC", compcode, reason, parms->qmname);

	/* indicate we no longer have a connection to the qm */
	qm = 0;
}

void connectQM(putParms * parms)

{
	MQLONG	compcode=0;
	MQLONG	reason=0;

	/* check for the same queue manager */
	if (strcmp(parms->qmname, connectName) != 0)
	{
		/* check if we have a connection */
		if (qm != 0)
		{
			discQM(parms);
		}
	}

	/* check if we are already connected to the queue manager */
	if (0 == qm)
	{
		/* Connect to the queue manager */
		printf("\nconnecting to queue manager %s\n",parms->qmname);
		MQCONN(parms->qmname, &qm, &compcode, &reason);

		/* check for errors */
		checkerror("MQCONN", compcode, reason, parms->qmname);
		if (compcode != 0)
		{
			err = 1;
		}

		/* remember the current qm */
		strcpy(connectName, parms->qmname);
	}
}

void closeQ(putParms * parms)

{
	MQLONG	compcode=0;
	MQLONG	reason=0;

	if (0 == q)
	{
		return;
	}

	/* close the input queue */
	printf("closing the queue\n");
	MQCLOSE(qm, &q, MQCO_NONE, &compcode, &reason);

	checkerror("MQCLOSE", compcode, reason, parms->qname);

	/* indicate q is no longer open */
	q = 0;
}

void openQ(putParms * parms)

{
	MQLONG	compcode=0;
	MQLONG	reason=0;
	MQLONG	openopt = 0;
	MQOD	objdesc = {MQOD_DEFAULT};

	/* check for the same queue */
	if ((strcmp(parms->qname, openName) != 0) || (strcmp(parms->remoteQM, openRemoteName) != 0))
	{
		/* check if we have an open queue */
		if (q != 0)
		{
			closeQ(parms);
		}
	}

	/* is the queue already open? */
	if (0 == q)
	{
		/* set the queue open options */
		strcpy(objdesc.ObjectName, parms->qname);
		strcpy(objdesc.ObjectQMgrName, parms->remoteQM);
		openopt = MQOO_OUTPUT + MQOO_FAIL_IF_QUIESCING;

		/* check if we need to set all context */
		if (MQMD_YES == parms->useMQMD)
		{
			openopt |= MQOO_SET_ALL_CONTEXT;
		}

		/* open the queue for output */
		printf("opening queue %s for output\n", parms->qname);
		MQOPEN(qm, &objdesc, openopt, &q, &compcode, &reason);

		/* check for errors */
		checkerror("MQOPEN", compcode, reason, parms->qname);
		if (compcode != 0)
		{
			err = 1;
		}

		/* remember the current queue name */
		strcpy(openName, parms->qname);
		strcpy(openRemoteName, parms->remoteQM);
	}
}

void commitQ(putParms * parms)

{
	MQLONG	compcode=0;
	MQLONG	reason=0;

	MQCMIT(qm, &compcode, &reason);
	checkerror("MQCMIT", compcode, reason, parms->qname);
}

/**************************************************************/
/*                                                            */
/* This routine puts a message on the queue.                  */
/*                                                            */
/**************************************************************/

int putMessage(char * msgdata, 
			   int msglen,
			   char * mqmdptr,
			   putParms * parms)

{
	MQLONG	compcode=0;
	MQLONG	reason=0;
	int		hasRFH;
	int		rfhlen;
	MQMD	msgdesc = {MQMD_DEFAULT};
	MQPMO	mqpmo = {MQPMO_DEFAULT};
	MQRFH	*tempRFH;
	char	tempid[5];

	/* check if we are using an mqmd from the file */
	if (mqmdptr != NULL)
	{
		/* set the get message options */
		if (uow > 0)
		{
			/* use syncpoints */
			mqpmo.Options = MQPMO_SYNCPOINT | MQPMO_FAIL_IF_QUIESCING;
		}
		else
		{
			/* no synchpoint, each message as a separate UOW */
			mqpmo.Options = MQPMO_NO_SYNCPOINT | MQPMO_FAIL_IF_QUIESCING;
		}

		mqpmo.Options |= MQPMO_SET_ALL_CONTEXT;

		if (1 == parms->newMsgId)
		{
			mqpmo.Options |= MQPMO_NEW_MSG_ID;
		}

		/* set the MQMD */
		memcpy(&msgdesc, mqmdptr, sizeof(MQMD));
	}
	else
	{
		/* set the get message options */
		if (uow > 0)
		{
			/* use syncpoints */
			mqpmo.Options = MQPMO_SYNCPOINT | MQPMO_NEW_MSG_ID | MQPMO_FAIL_IF_QUIESCING;
		}
		else
		{
			/* no synchpoint, each message as a separate UOW */
			mqpmo.Options = MQPMO_NO_SYNCPOINT | MQPMO_NEW_MSG_ID | MQPMO_FAIL_IF_QUIESCING;
		}

		if ((1 == parms->inGroup) || (1 == parms->lastGroup))
		{
			mqpmo.Options |= MQPMO_LOGICAL_ORDER;

			if (1 == parms->inGroup)
			{
				msgdesc.MsgFlags |= MQMF_MSG_IN_GROUP;
			}

			if (1 == parms->lastGroup)
			{
				msgdesc.MsgFlags |= MQMF_LAST_MSG_IN_GROUP;
			}
		}

		/* Indicate V2 of MQMD */
		msgdesc.Version = MQMD_VERSION_2;

		/* set the persistence, etc if specified */
		msgdesc.Persistence = parms->persist;
		msgdesc.Encoding = parms->encoding;
		msgdesc.CodedCharSetId = parms->codepage;

		/* check if message expiry was specified */
		if (parms->expiry > 0)
		{
			msgdesc.Expiry = parms->expiry;
		}

		/* check if message type was specified */
		if (parms->msgtype > 0)
		{
			msgdesc.MsgType = parms->msgtype;
		}

		/* check if message priority was specified */
		if (parms->priority != MQPRI_PRIORITY_AS_Q_DEF)
		{
			msgdesc.Priority = parms->priority;
		}

		/* check if report options were specified */
		if (parms->report > 0)
		{
			msgdesc.Report = parms->report;
		}

		/* set the message format in the MQMD was specified */
		switch (parms->rfh)
		{
		case RFH_NO:
			{
				if (1 == parms->formatSet)
				{
					memcpy(msgdesc.Format, parms->msgformat, MQ_FORMAT_LENGTH);
				}

				break;
			}
		case RFH_V1:
			{
				memcpy(msgdesc.Format, MQFMT_RF_HEADER, sizeof(msgdesc.Format));
				break;
			}
		case RFH_V2:
			{
				memcpy(msgdesc.Format, MQFMT_RF_HEADER_2, sizeof(msgdesc.Format));
				break;
			}
		case 'A':
			{
				/* check if the message data contains an rfh header */
				hasRFH = 0;
				if (msglen >= sizeof(MQRFH))
				{
					/* translate the first 4 bytes from EBCDIC to ASCII */
					memset(tempid, 0, sizeof(tempid));
					EbcdicToAscii((unsigned char *) msgdata, sizeof(MQRFH_STRUC_ID) - 1, (unsigned char *) &tempid);

					/* check for either an ASCII or EBCDIC structure id */
					if ((memcmp(tempid, MQRFH_STRUC_ID, sizeof(MQRFH_STRUC_ID) - 1) == 0) ||
						(memcmp(msgdata, MQRFH_STRUC_ID, sizeof(MQRFH_STRUC_ID) - 1) == 0))
					{
						/* the first four bytes match an RFH structure id */
						tempRFH = (MQRFH *) msgdata;
						hasRFH = tempRFH->Version;
						rfhlen = tempRFH->StrucLength;
						
						/* check for the RFH version field in PC format */
						if ((MQRFH_VERSION_1 != hasRFH) && (MQRFH_VERSION_2 != hasRFH))
						{
							/* reverse the order, and check for host integers */
							hasRFH = reverseBytes4(hasRFH);
							rfhlen = reverseBytes4(rfhlen);

							/* check for the RFH version field in host format */
							if ((MQRFH_VERSION_1 == hasRFH) || (MQRFH_VERSION_2 == hasRFH))
							{
								/* host encoding - check the length */
								if ((rfhlen < sizeof(MQRFH)) || (rfhlen > msglen))
								{
									/* invalid length - do not treat as an RFH header */
									hasRFH = 0;
								}
							}
							else
							{
								/* do not recognize the version */
								hasRFH = 0;
							}
						}
						else
						{
							/* normal encoding - check the rfh length */
							if ((rfhlen < sizeof(MQRFH)) || (rfhlen > msglen))
							{
								/* invalid length */
								hasRFH = 0;
							}
						}
					}

				}

				switch (hasRFH)
				{
				case 1:
					{
						memcpy(msgdesc.Format, MQFMT_RF_HEADER, sizeof(msgdesc.Format));
						break;
					}
				case 2:
					{
						memcpy(msgdesc.Format, MQFMT_RF_HEADER_2, sizeof(msgdesc.Format));
						break;
					}

				}

				break;
			}
		case RFH_XML:
			{
				memcpy(msgdesc.Format, MQFMT_RF_HEADER_2, sizeof(msgdesc.Format));
				break;
			}
		}

		/* check if a reply to queue manager was specified */
		memset(msgdesc.ReplyToQMgr, 0, sizeof(msgdesc.ReplyToQMgr));
		if (parms->replyQM[0] != 0)
		{
			memcpy(msgdesc.ReplyToQMgr, parms->replyQM, strlen(parms->replyQM));
		}

		/* check if a reply to queue was specified */
		memset(msgdesc.ReplyToQ, 0, sizeof(msgdesc.ReplyToQ));
		if (parms->replyQ[0] != 0)
		{
			memcpy(msgdesc.ReplyToQ, parms->replyQ, strlen(parms->replyQ));
		}

		/* check if a correl id was specified */
		if (parms->correlidSet == 1)
		{
			memcpy(msgdesc.CorrelId, parms->correlid, MQ_CORREL_ID_LENGTH);
		}
		else
		{
			memset(msgdesc.CorrelId, 0, MQ_CORREL_ID_LENGTH);
		}

		/* check if a group id was specified */
		if (1 == groupOpen)
		{
			memcpy(msgdesc.GroupId, parms->saveGroupId, MQ_GROUP_ID_LENGTH);
		}
		else
		{
			if (1 == parms->groupidSet)
			{
				memcpy(msgdesc.GroupId, parms->groupid, MQ_GROUP_ID_LENGTH);
				msgdesc.MsgFlags |= MQMF_LAST_MSG_IN_GROUP | MQMF_MSG_IN_GROUP ;
			}
			else
			{
				memset(msgdesc.GroupId, 0, sizeof(msgdesc.GroupId));
			}
		}
	}

	/* perform the MQPUT */
	MQPUT(qm, q, &msgdesc, &mqpmo, msglen, msgdata, &compcode, &reason);

	/* check if we got to the end of the queue */
	/* check for errors */
	checkerror("MQPUT", compcode, reason, parms->qname);

	/* check if this message is part of a group */
	if (1 == parms->inGroup)
	{
		groupOpen = 1;
	}

	/* check if this is the last message in a group */
	if (1 == parms->lastGroup)
	{
		groupOpen = 0;
	}

	if (1 == groupOpen)
	{
		memset(parms->saveGroupId, 0, sizeof(parms->saveGroupId));
		memcpy(parms->saveGroupId, msgdesc.GroupId, MQ_GROUP_ID_LENGTH);
	}

	memcpy(puttime, msgdesc.PutTime, sizeof(msgdesc.PutTime));

	return compcode;
}

/**************************************************************/
/*                                                            */
/* Subroutine to format a time.                               */
/*                                                            */
/* The input time should be 8 numbers (hhmmsshh)              */
/*                                                            */
/**************************************************************/

void formatTime(char *timeOut, char *timeIn)

{
	timeOut[0] = timeIn[0];
	timeOut[1] = timeIn[1];
	timeOut[2] = ':';
	timeOut[3] = timeIn[2];
	timeOut[4] = timeIn[3];
	timeOut[5] = ':';
	timeOut[6] = timeIn[4];
	timeOut[7] = timeIn[5];
	timeOut[8] = ':';
	timeOut[9] = timeIn[6];
	timeOut[10] = timeIn[7];
	timeOut[11] = 0;
}

void processMessageFile(const char * fileName, putParms * parms)

{
	MQLONG	cc=MQCC_OK;
	MQLONG	msglen=0;
	int		rc;
	char *	mqmdptr=NULL;
	char *	buffer=NULL;
	char *	ptr;

	/* pointer to data file areas */
	int		filedatalen=0;
	char *	filedata=NULL;
	char *	filedataOfs=NULL;
	char	formTime[16];

	memset(puttime, 0, sizeof(puttime));

	/* process this line as a message data file name */
	/* read the message data file */
	rc = readFileData(fileName, &filedatalen, &filedata, parms);

	/* check for delimiters in the data */
	if ((0 == rc) && (filedata != NULL))
	{
		/* check if we have a connection to the queue manager */
		connectQM(parms);

		/* check if the queue has been opened */
		openQ(parms);

		if (filedatalen > 0)
		{
			buffer = filedata;

			while ((0 == err) && (filedatalen > 0))
			{
				ptr = scanForDelim(buffer, filedatalen, parms);

				if (ptr != NULL)
				{
					/* found a delimiter in the file data */
					/* shorten the data length */
					/* get the length of this message */
					msglen = ptr - buffer;

				}
				else
				{
					msglen = filedatalen;
				}

				mqmdptr = NULL;

				if (1 == parms->useMQMD)
				{
					/* check for the presence of an MQMD */
					if ((msglen > sizeof(MQMD)) && (memcmp(buffer, MQMD_STRUC_ID, 4) == 0))
					{
						mqmdptr = buffer;
						buffer += sizeof(MQMD);
						msglen -= sizeof(MQMD);
						filedatalen -= sizeof(MQMD);
					}
				}

				/* put the data to the queue */
				cc = putMessage(buffer,
								msglen + parms->rfhlength,
								mqmdptr,
								parms);

				if (MQCC_OK == cc)
				{
					if (0 == parms->msgwritten)
					{
						/* write out the time of the first message */
						formatTime(formTime, puttime);
						printf("first message written at %8.8s\n", formTime);
					}

					/* increment the message count */
					parms->msgwritten++;
					parms->byteswritten += msglen;

					if (0 == groupOpen)
					{
						commitQ(parms);
					}
					/* check if any thinktime has been specified */
					if (parms->thinkTime > 0)
					{
						Sleep(parms->thinkTime);
					}
				}


				/* update the data pointer */
				buffer += (msglen + parms->delimiterLen);

				/* calculate the remaining bytes */
				filedatalen -= (msglen + parms->delimiterLen);
			}
		}
		else
		{
			/* put one zero length message */
			/* since there might be an RFH */
		}
	}
	else
	{
		if (rc != 0)
		{
			printf("Error reading file %s - rc %d\n", fileName, rc);
		}
	}

	if (filedata != NULL)
	{
		free(filedata);
	}
}

void procParmFile(FILE * parmfile, putParms * parms)

{
	int		len;
	char *	ptr;
	char	parmline[512];

	/* read the parameter file */
	while ((0 == err) && (fgets(parmline, sizeof(parmline) - 1, parmfile) != NULL))
	{
		/* check for a new line character at the end of the line */
		while ((strlen(parmline) > 0) && (parmline[strlen(parmline) - 1] < ' '))
		{
			/* get rid of the new line character */
			parmline[strlen(parmline) - 1] = 0;
		}

		/* point to the beginning of the line */
		/* skip any leading blanks */
		ptr = skipBlanks(parmline);

		/* truncate any trailing blanks */
		len = strlen(ptr);
		while ((len > 0) && (ptr[len] == ' '))
		{
			ptr[len] = 0;
			len--;
		}

		/* check for a comment or blank line */
		if ((ptr[0] != 0) && (ptr[0] != ';') && (ptr[0] != '#') && (ptr[0] != '*'))
		{
			if ('<' == ptr[0])
			{
				/* process this as an XML data entry */
				if (0 == foundUsr)
				{
					/* must be the first entry */
					foundUsr = processFirstUsrLine(ptr, parms);
				}
				else
				{
					/* figure out what kind of entry this belongs to */
					foundUsr = processUsrLine(ptr, foundUsr, parms);
				}
			}
			else
			{
				if ((0 == foundfiles) || ('[' == ptr[0]))
				{
					/* process this line as message parameters */
					foundfiles = processParmLine(ptr, parms);
				}
				else
				{
					processMessageFile(ptr, parms);
				}
			}
		}

		/* re-initialize the parameter input area */
		memset(parmline, 0, sizeof(parmline));
	}
}

/**************************************************************/
/*                                                            */
/* Display command format.                                    */
/*                                                            */
/**************************************************************/

void printHelp(char *pgmName)

{
	printf("%s\n", Level);
	printf("format is:\n");
	printf("  %s -f parm_file {-v}\n", pgmName);
	printf("   parm_file is the fully qualified name of the parameters file\n");
	printf("   -v verbose\n");
}

int main(int argc, char **argv)

{
	FILE *		parmfile;
	char		formTime[16];
	putParms	parms;

	printf(Level);

	/* print the copyright statement */
	printf(copyright);

	/* initialize the work areas */
	memset(&parms, 0, sizeof(putParms));
	memset(formTime, 0, sizeof(formTime));

	/* initialize the open and connection names */
	memset(openName, ' ', sizeof(openName));
	memset(connectName, ' ', sizeof(connectName));

	/* check for too few input parameters */
	if (argc < 2)
	{
		printHelp(argv[0]);
		exit(99);
	}

	/* check for help request */
	if ((argv[1][0] == '?') || (argv[1][1] == '?'))
	{
		printHelp(argv[0]);
		exit(0);
	}

	/* process any command line arguments */
	processArgs(argc, argv, &parms);

	if (err != 0)
	{
		printHelp(argv[0]);
		exit(99);
	}

	if (0 == parms.parmFilename[0])
	{
		printf("Parameters file name not specified - use -f parameter to specify a parameters file\n");
		printHelp(argv[0]);
		exit(0);
	}

	/* open the parameters file */
	parmfile = fopen(parms.parmFilename, "r");

	if (parmfile != NULL)
	{
		/* start to process the parameters file a message at a time */
		procParmFile(parmfile, &parms);

		fclose(parmfile);
	}
	else
	{
		printf("Unable to open parameters file %s\n", parms.parmFilename);
	}

	if (err != 0)
	{
		printf("***** Errors were encountered during the run\n");
	}

	if (parms.msgwritten > 0)
	{
		/* write out the time of the last message */
		formatTime(formTime, puttime);
		printf("last message written at %8.8s\n", formTime);

		/* dump out the total message count */
#ifdef WIN32
		printf("Total messages written %I64d\n", parms.msgwritten);
		printf("Total bytes written    %I64d\n", parms.byteswritten);
#else
		printf("Total messages written %lld\n", parms.msgwritten);
		printf("Total bytes written    %lld\n", parms.byteswritten);
#endif
	}
	else
	{
		printf("No messages written - check parameters file\n");
	}

	/* close any open queues */
	closeQ(&parms);

	/* disconnect from the qm */
	discQM(&parms);

	/******************************************************************/
	/*                                                                */
	/* END OF PROGRAM                                                 */
	/*                                                                */
	/******************************************************************/

	printf("MQTEST program ended\n");

	return(0);
}
