cl_msg.c

linux集群服务器软件代码包

	const void	*ret;
	int		type;
	ret = cl_get_value( msg, name, NULL, &type);

	if (ret == NULL || type != FT_STRING){
		return(NULL);
	}

	return(ret);

}

int
cl_get_type(const struct ha_msg *msg, const char *name)
{

	const void	*ret;
	int		type;

	ret =  cl_get_value( msg, name, NULL, &type);

	if (ret == NULL) {
		return -1;
	}
	if (type < 0){
		cl_log(LOG_WARNING, "field %s not a valid type"
		       ,	name);
		return(-1);
	}

	return(type);

}

struct ha_msg *
cl_get_struct(const struct ha_msg *msg, const char* name)
{
	struct ha_msg	*ret;
	int		type;
	size_t		vallen;

	ret = (struct ha_msg *)cl_get_value(msg, name, &vallen, &type);
	
	if (ret == NULL || type != FT_STRUCT){
		return(NULL);
	}
	return(ret);
}

int
cl_msg_list_length(struct ha_msg* msg, const char* name)
{
	GList*   ret;
	int		type;
	
	ret = cl_get_value( msg, name, NULL, &type);
	
	if ( ret == NULL || type != FT_LIST){
		return -1;
	}

	return g_list_length(ret);
	
}


void* 
cl_msg_list_nth_data(struct ha_msg* msg, const char* name, int n)
{
	GList*   ret;
	int		type;
	
	ret = cl_get_value( msg, name, NULL, &type);
	
	if ( ret == NULL || type != FT_LIST){
		cl_log(LOG_WARNING, "field %s not found "
		       " or type mismatch", name);
		return NULL;
	}
	
	return g_list_nth_data(ret, n);
	
}


static int
cl_msg_mod(struct ha_msg * msg, const char * name,
	       const void* value, size_t vlen, int type)
{  
  	int j;
	
	AUDITMSG(msg);
	if (msg == NULL || name == NULL || value == NULL) {
		cl_log(LOG_ERR, "cl_msg_mod: NULL input.");
		return HA_FAIL;
	}
	
	if(type >= DIMOF(fieldtypefuncs)){
		cl_log(LOG_ERR, "cl_msg_mod:"
		       "invalid type(%d)", type);
		return HA_FAIL;
	}

	for (j=0; j < msg->nfields; ++j) {
		if (strcmp(name, msg->names[j]) == 0) {
			
			char *	newv ;
			int	newlen = vlen;
			int	string_sizediff = 0;
			int	netstring_sizediff = 0;
			
			newv = fieldtypefuncs[type].dup(value,vlen);
			if (!newv){
				cl_log(LOG_ERR, "dupliationg message fields failed"
				       "value=%p, vlen=%d, msg->names[j]=%s", 
				       value, (int)vlen, msg->names[j]);
				return HA_FAIL;
			}
			
			if (type != FT_STRUCT){
				string_sizediff = fieldtypefuncs[type].stringlen(strlen(name), vlen, value)
					- fieldtypefuncs[type].stringlen(strlen(name), 
									 msg->vlens[j], msg->values[j]);
				netstring_sizediff = fieldtypefuncs[type].netstringlen(strlen(name), vlen, value)
					- fieldtypefuncs[type].netstringlen(strlen(name), 
									    msg->vlens[j], msg->values[j]);
				msg->stringlen += string_sizediff;
				msg->netstringlen += netstring_sizediff;
			}
			
			
			fieldtypefuncs[type].memfree(msg->values[j]);
			msg->values[j] = newv;
			msg->vlens[j] = newlen;
			AUDITMSG(msg);
			return(HA_OK);
		}
	}
	
	return(ha_msg_nadd_type(msg, name,strlen(name), value, vlen, type));
  
}

int
cl_msg_modstruct(struct ha_msg * msg, const char* name, 
		 const struct ha_msg* value)
{
	return cl_msg_mod(msg, name, value, 0, FT_STRUCT);	
}

int
cl_msg_modbin(struct ha_msg * msg, const char* name, 
	      const void* value, size_t vlen)
{
	return cl_msg_mod(msg, name, value, vlen, FT_BINARY);
	
}
int
cl_msg_moduuid(struct ha_msg * msg, const char* name, 
	       const uuid_t uuid)
{
	return cl_msg_mod(msg, name, uuid, sizeof(uuid_t), FT_BINARY);
}
	


/* Modify the value associated with a particular name */
int
cl_msg_modstring(struct ha_msg * msg, const char * name, const char * value)
{
	return cl_msg_mod(msg, name, value, strlen(value), FT_STRING);
}



/* Return the next message found in the stream */
struct ha_msg *
msgfromstream(FILE * f)
{
	char		buf[MAXMSGLINE];
	char *		getsret;
	clearerr(f);
	/* Skip until we find a MSG_START (hopefully we skip nothing) */
	while(1) {
		getsret = fgets(buf, sizeof(buf), f);
		if (!getsret) {
			break;
		}
		if (strcmp(buf, MSG_START) == 0) {
			return msgfromstream_string(f);

		}
		if (strcmp(buf, MSG_START_NETSTRING) == 0){
			return msgfromstream_netstring(f);
		}

	}

	return NULL;
}

/* Return the next message found in the stream with string format */
struct ha_msg *
msgfromstream_string(FILE * f)
{
	char		buf[MAXMSGLINE];
	const char *	bufmax = buf + sizeof(buf);
	struct ha_msg*	ret;
	char *		getsret;


	if ((ret = ha_msg_new(0)) == NULL) {
		/* Getting an error with EINTR is pretty normal */
		/* (so is EOF) */
		if (   (!ferror(f) || (errno != EINTR && errno != EAGAIN))
		&&	!feof(f)) {
			cl_log(LOG_ERR, "msgfromstream: cannot get message");
		}
		return(NULL);
	}

	/* Add Name=value pairs until we reach MSG_END or EOF */
	while(1) {
		getsret = fgets(buf, MAXMSGLINE, f);
		if (!getsret) {
			break;
		}

		if (strnlen(buf, MAXMSGLINE) > MAXMSGLINE - 2) {
			cl_log(LOG_DEBUG
			,	"msgfromstream: field too long [%s]"
			,	buf);
		}

		if (!strcmp(buf, MSG_END)) {
			break;
		}


		/* Add the "name=value" string on this line to the message */
		if (ha_msg_add_nv(ret, buf, bufmax) != HA_OK) {
			cl_log(LOG_ERR, "NV failure (msgfromsteam): [%s]"
			,	buf);
			ha_msg_del(ret); ret=NULL;
			return(NULL);
		}
	}
	return(ret);
}


/* Return the next message found in the stream with string format*/

struct ha_msg *
msgfromstream_netstring(FILE * f)
{
	struct ha_msg *		ret;
	char			total_databuf[MAXMSG];
	char *			sp = total_databuf;
	int (*netstringtofield)(const void*, size_t, void**, size_t*);			
	void (*memfree)(void*);
	

	if ((ret = ha_msg_new(0)) == NULL) {
		/* Getting an error with EINTR is pretty normal */
		/* (so is EOF) */
		if (   (!ferror(f) || (errno != EINTR && errno != EAGAIN))
		&&	!feof(f)) {
			cl_log(LOG_ERR
			, "msgfromstream_netstring(): cannot get message");
		}
		return(NULL);
	}

	while(1) {
		int	namelen=-1;
		char *	name;
		char *	namebuf;
		int	datalen;
		char *	data;
		char *	databuf;
		int	n;
		int	typelen;
		char *  type;
		char *	typebuf;
		int		fieldtype;
		void*		ret_data;
		size_t		ret_datalen;
		
		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(LOG_INFO, 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);
		
				
		fieldtype = atoi(type);


		
		if (fieldtype < DIMOF(fieldtypefuncs)){					
			netstringtofield = fieldtypefuncs[fieldtype].netstringtofield;
			memfree = fieldtypefuncs[fieldtype].memfree;
			if (!netstringtofield || netstringtofield(data, datalen,
								  &ret_data, &ret_datalen) != HA_OK){
				cl_log(LOG_ERR, "msgfromstream_netstring: "
				       "tonetstring() failed, type=%d", fieldtype);
				return NULL;
			}
		}else {
			cl_log(LOG_ERR, "msgfromstream_netstring: wrong type(%d)", fieldtype);
			return NULL;
		}
		
		if (ha_msg_nadd_type(ret, name, namelen, data, datalen, fieldtype)
		    != HA_OK) {
			cl_log(LOG_ERR, "ha_msg_nadd fails(netstring2msg)");
			
			if (memfree && ret_data){
				memfree(ret_data);
			} else{
				cl_log(LOG_ERR, "netstring2msg:"
				       "memfree or ret_value is NULL");
			}
			
			ha_msg_del(ret);
			return(NULL);
		}
		
		
		if (memfree && ret_data){
			memfree(ret_data);
		} else{
			cl_log(LOG_ERR, "netstring2msg:"
			       "memfree or ret_value is 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);
}