cl_msg.c

在LINUX下实现HA的源代码

	}

	while(1) {
		int	namelen=-1;
		char *	name;
		char *	namebuf;
		int	datalen;
		char *	data;
		char *	databuf;
		int	n;
		int	typelen;
		char *  type;
		char *	typebuf;

		if (fscanf(f, "%d:", &namelen) <= 0 || namelen <= 0){
			if (!cl_msg_quiet_fmterr) {
				cl_log(LOG_WARNING
				,	" msgfromstream_netstring()"
				": scanning for namelen failed");
			}
			ha_msg_del(ret);
			return(NULL);
		}

		namebuf = ha_malloc(namelen + 2);

		if ((n = fread(namebuf, 1, namelen + 1, f)) != namelen + 1){
			cl_log(LOG_WARNING, "msgfromstream_netstring()"
			": Can't get enough name string,"
			"expecting %d bytes long name, got %d bytes"
			,	namelen, n);
			ha_msg_del(ret);
			return(NULL);
		}

		if (*(namebuf + namelen) != ',' ){
			if (!cl_msg_quiet_fmterr) {
				cl_log(LOG_WARNING
				,	"msgfromstream_netstring()"
				": \",\" is missing in netstring for name");
			}
			ha_msg_del(ret);
			return(NULL);
		}

		namebuf[namelen] = 0;
		name = namebuf;

		if (fscanf(f, "%d:", &typelen) <= 0 || typelen <= 0){

			if (!is_auth_netstring(total_databuf
			,	sp - total_databuf, name,namelen) ){
				if (!cl_msg_quiet_fmterr) {
					cl_log(LOG_ERR
					,	"msgfromstream_netstring()"
					": netstring authentication"
					" failed msgfromstream_netstring()");
				}
				cl_log_message(ret);
				ha_msg_del(ret);
				return(NULL);
			}

			return(ret);
		}

		typebuf = ha_malloc(typelen + 2);
		if ((n = fread(typebuf, 1, typelen + 1, f)) != typelen + 1){
			cl_log(LOG_WARNING
			,	"msgfromstream_netstring()"
			": Can't get enough type string,"
			"expecting %d bytes long type, got %d type"
			,	typelen, n);
			ha_msg_del(ret);
			return(NULL);
		}

		if (*(typebuf + typelen) != ',' ){
			if  (!cl_msg_quiet_fmterr) {
				cl_log(LOG_WARNING
				,	"msgfromstream_netstring()"
				": \",\" is missing in netstring for type");
			}
			ha_msg_del(ret);
			return(NULL);
		}

		typebuf[typelen] = 0;
		type = typebuf;

		if (fscanf(f, "%d:", &datalen) <= 0) {
			if (!cl_msg_quiet_fmterr) {
				cl_log(LOG_WARNING
				,	"msgfromstream_netstring()"
				": scanning for datalen failed");
			}
			ha_msg_del(ret);
			return(NULL);
		}

		databuf = ha_malloc(datalen + 2);

		if ((n = fread(databuf, 1, datalen + 1, f)) != datalen + 1) {
			cl_log(LOG_WARNING
			,	"msgfromstream_netstring()"
			": Can't get enough data"
			", expecting %d bytes long data, got %d bytes"
			,	datalen, n);
			ha_msg_del(ret);
			return(NULL);
		}

		if (*(databuf + datalen ) != ',' ){
			if (!cl_msg_quiet_fmterr) {
				cl_log(LOG_WARNING
				,	"msgfromstream_netstring()"
				": \",\" is missing in netstring for data");
			}
			ha_msg_del(ret);
			return(NULL);
		}

		databuf[datalen] = 0;
		data = databuf ;

		sp += sprintf(sp, "%d:%s,", namelen, name);
		sp += sprintf(sp, "%d:%s,", typelen, type);
		sp += sprintf(sp, "%d:%s,", datalen, data);

		if (atoi(type) == FT_STRUCT){
			struct ha_msg	*tmpmsg;

			tmpmsg = netstring2msg(data, datalen, 1);
			data = (char*)tmpmsg;
			datalen = sizeof(struct ha_msg);
		}

		if (ha_msg_nadd_type(ret, name, namelen, data, datalen
		,	atoi(type)) != HA_OK){
			cl_log(LOG_WARNING
			,  "msgfromstream_netstring(): ha_msg_nadd_type fails");
			ha_msg_del(ret);
			return(NULL);
		}

		ha_free(namebuf);
		ha_free(databuf);
	}

}




/* Return the next message found in the IPC channel */
static struct ha_msg*
msgfromIPC_ll(IPC_Channel * ch, int need_auth)
{
	int		rc;
	IPC_Message*	ipcmsg;
	struct ha_msg*	hmsg;

	rc = ch->ops->waitin(ch);

	switch(rc) {
		default:
		case IPC_FAIL:
			cl_perror("msgfromIPC: waitin failure");
			return NULL;

		case IPC_BROKEN:
			sleep(1);
			return NULL;

		case IPC_INTR:
			return NULL;

		case IPC_OK:
			break;
	}


	ipcmsg = NULL;
	rc = ch->ops->recv(ch, &ipcmsg);
#if 0
	if (DEBUGPKTCONT) {
		cl_log(LOG_DEBUG, "msgfromIPC: recv returns %d ipcmsg = 0x%lx"
		,	rc, (unsigned long)ipcmsg);
	}
#endif
	if (rc != IPC_OK) {
		return NULL;
	}


	hmsg = wirefmt2msg_ll((char *)ipcmsg->msg_body, ipcmsg->msg_len, need_auth);
	if (ipcmsg->msg_done) {
		ipcmsg->msg_done(ipcmsg);
	}

	AUDITMSG(hmsg);
	return hmsg;
}



/* Return the next message found in the IPC channel */
struct ha_msg*
msgfromIPC(IPC_Channel * ch)
{
	return msgfromIPC_ll(ch, 1);
}


struct ha_msg*
msgfromIPC_noauth(IPC_Channel * ch)
{
	return msgfromIPC_ll(ch, 0);
}





/* Return the next message found in the IPC channel */
IPC_Message *
ipcmsgfromIPC(IPC_Channel * ch)
{
	int		rc;
	IPC_Message*	ipcmsg;

	rc = ch->ops->waitin(ch);

	switch(rc) {
		default:
		case IPC_FAIL:
			cl_perror("msgfromIPC: waitin failure");
			return NULL;

		case IPC_BROKEN:
			sleep(1);
			return NULL;

		case IPC_INTR:
			return NULL;

		case IPC_OK:
			break;
	}


	ipcmsg = NULL;
	rc = ch->ops->recv(ch, &ipcmsg);
#if 0
	if (DEBUGPKTCONT) {
		cl_log(LOG_DEBUG, "msgfromIPC: recv returns %d ipcmsg = 0x%lx"
		,	rc, (unsigned long)ipcmsg);
	}
#endif
	if (rc != IPC_OK) {
		return NULL;
	}


	return(ipcmsg);
}


/* Writes a message into a stream - used for serial lines */
int
msg2stream(struct ha_msg* m, FILE * f)
{
	size_t	len;
	char *	s  = msg2wirefmt(m, &len);

	if (s != NULL) {
		int	rc = HA_OK;
		if (fputs(s, f) == EOF) {
			rc = HA_FAIL;
			cl_perror("msg2stream: fputs failure");
		}
		if (fflush(f) == EOF) {
			cl_perror("msg2stream: fflush failure");
			rc = HA_FAIL;
		}
		ha_free(s);
		return(rc);
	}else{
		return(HA_FAIL);
	}
}
static void ipcmsg_done(IPC_Message* m);

static void
ipcmsg_done(IPC_Message* m)
{
	if (!m) {
		return;
	}
	if (m->msg_body) {
		ha_free(m->msg_body);
	}
	ha_free(m);
	m = NULL;
}


IPC_Message*
wirefmt2ipcmsg(void* p, size_t len, IPC_Channel* ch)
{
	IPC_Message*	ret = NULL;

	if (p == NULL){
	  return(NULL);
	}

	ret = MALLOCT(IPC_Message);
	if (!ret) {
		return(NULL);
	}
	ret->msg_done = ipcmsg_done;
	ret->msg_private = NULL;
	ret->msg_ch = ch;
	ret->msg_body = p;
	ret->msg_len = len;

	return ret;

}

IPC_Message*
hamsg2ipcmsg(struct ha_msg* m, IPC_Channel* ch)
{
	size_t		len;
	char *		s  = msg2wirefmt(m, &len);
	IPC_Message*	ret = NULL;



	if (s == NULL) {
		return ret;
	}
	ret = MALLOCT(IPC_Message);
	if (!ret) {
		ha_free(s);
		return ret;
	}
	ret->msg_done = ipcmsg_done;
	ret->msg_private = NULL;
	ret->msg_ch = ch;
	ret->msg_body = s;
	ret->msg_len = len;

	return ret;
}

struct ha_msg*
ipcmsg2hamsg(IPC_Message*m)
{
	struct ha_msg*	ret = NULL;


	ret = wirefmt2msg(m->msg_body, m->msg_len);
	return ret;
}

int
msg2ipcchan(struct ha_msg*m, IPC_Channel*ch)
{
	IPC_Message*	imsg;

	if (m == NULL || ch == NULL) {
		cl_log(LOG_ERR, "Invalid msg2ipcchan argument");
		errno = EINVAL;
		return HA_FAIL;
	}

	if ((imsg = hamsg2ipcmsg(m, ch)) == NULL) {
		cl_log(LOG_ERR, "hamsg2ipcmsg() failure");
		return HA_FAIL;
	}

	if (ch->ops->send(ch, imsg) != IPC_OK) {
		if (ch->ch_status == IPC_CONNECT) {
			cl_log(LOG_ERR
			,	"msg2ipcchan: ch->ops->send() failure");
		}
		imsg->msg_done(imsg);
		return HA_FAIL;
	}
	return HA_OK;
}

static gboolean (*msg_authentication_method)(const struct ha_msg* ret) = NULL;


void
cl_set_oldmsgauthfunc(gboolean (*authfunc)(const struct ha_msg*))
{
	msg_authentication_method = authfunc;
}


/* Converts a string (perhaps received via UDP) into a message */
static struct ha_msg *
string2msg_ll(const char * s, size_t length, int depth, int need_auth)
{
	struct ha_msg*	ret;
	int		startlen;
	int		endlen;
	const char *	sp = s;
	const char *	smax = s + length;


	if ((ret = ha_msg_new(0)) == NULL) {
		return(NULL);
	}

	startlen = sizeof(MSG_START)-1;
	if (strncmp(sp, MSG_START, startlen) != 0) {
		/* This can happen if the sender gets killed */
		/* at just the wrong time... */
		if (!cl_msg_quiet_fmterr) {
			cl_log(LOG_WARNING, "string2msg_ll: no MSG_START");
		}
		ha_msg_del(ret);
		return(NULL);
	}else{
		sp += startlen;
	}

	endlen = sizeof(MSG_END)-1;

	/* Add Name=value pairs until we reach MSG_END or end of string */

	while (*sp != EOS && strncmp(sp, MSG_END, endlen) != 0) {

		if (sp >= smax)		return(NULL);
		/* Skip over initial CR/NL things */
		sp += strspn(sp, CRNL);
		if (sp >= smax)		return(NULL);

		/* End of message marker? */
		if (strncmp(sp, MSG_END, endlen) == 0) {
			break;
		}
		/* Add the "name=value" string on this line to the message */
		if (ha_msg_add_nv_depth(ret, sp, smax, depth) != HA_OK) {
			if (!cl_msg_quiet_fmterr) {
				cl_log(LOG_ERR, "NV failure (string2msg_ll):");
				cl_log(LOG_ERR, "Input string: [%s]", s);
			}
			ha_msg_del(ret);
			return(NULL);
		}
		if (sp >= smax) {
			return(NULL);
		}
		sp += strcspn(sp, CRNL);
	}

	if (need_auth && msg_authentication_method
	&&		!msg_authentication_method(ret)) {
		const char* from = ha_msg_value(ret, F_ORIG);
		if (!cl_msg_quiet_fmterr) {
			cl_log(LOG_WARNING
		       ,       "string2msg_ll: node [%s]"
		       " failed authentication", from ? from : "?");
		}
		ha_msg_del(ret);
		ret = NULL;
	}
	return(ret);
}

struct ha_msg *
string2msg(const char * s, size_t length)
{
	return(string2msg_ll(s, length, 0, NEEDAUTH));
}


/* Converts a message into a string (for sending out UDP interface)

   used in two places:

   1.called by msg2string as a implementation for computing string for a
   message provided the buffer

   2.called by is_authentic. In this case, there are no start/end string
   and the "auth" field is not included in the string

   rules for generating a string:

   1) if the field is a string, then add "name=value" in the string followed by
   new line

   2) if the field is binary data, then add "(FT_BINARY)name=
   base64-version-of-binary-data" followed by a new line

   3) if the field is a child message, then add "(FT_STRUCT)name=
   converted-string-for-child-message" followed by a new line


*/


int
msg2string_buf(const struct ha_msg *m, char* buf, size_t len
,	int depth,int needhead)
{

	char *	bp = NULL;
	int	j;

	buf[0]=0;
	bp = buf;

	if (needhead){
		strcpy(bp, MSG_START);
		bp += strlen(MSG_START);
	}

	for (j=0; j < m->nfields; ++j) {

		if (needhead == NOHEAD && strcmp(m->names[j], F_AUTH) == 0) {
			continue;
		}

		if (m->types[j] == FT_BINARY || m->types[j] == FT_STRUCT){
			strcat(bp, "(");
			bp++;
			strcat(bp,FT_strings[m->types[j]]);
			bp++;
			strcat(bp,")");
			bp++;
		}

		strcat(bp, m->names[j]);
		bp += m->nlens[j];
		strcat(bp, "=");
		bp++;

		if (m->types[j] == FT_STRING ){

			strcat(bp, m->values[j]);
			bp += m->vlens[j];

		} else if (m->types[j] == FT_BINARY){
			int baselen;
			int truelen = 0;

			baselen = B64_stringlen(m->vlens[j]) + 1;
			truelen = binary_to_base64(m->values[j]
			,	m->vlens[j], bp, baselen);
			bp += truelen;
		} else{
			int	baselen = get_stringlen(
			(struct ha_msg*)	m->values[j], 0);

			if (msg2string_buf((struct ha_msg*)m->values[j]
			,	bp,baselen,depth + 1, NEEDHEAD) != HA_OK){

				cl_log(LOG_ERR
				, "msg2string_buf(): msg2string_buf for"
				" child message failed");