static int processNetflowV9Packet(IpfixReceiver* ipfixReceiver, byte* message, uint16_t length) { NetflowV9Header* header = (NetflowV9Header*)message;   IpfixSetHeader* set = (IpfixSetHeader*)&header->data;  int i; for (i = 0; i < ntohs(header->setCount); i++) { if (ntohs(set->id) == NetflowV9_SetId_Template) { processTemplateSet(ipfixReceiver, ntohs(header->sourceId), set); } else if (ntohs(set->id) >= IPFIX_SetId_Data_Start) { processDataSet(ipfixReceiver, ntohs(header->sourceId), set); } else { msg(MSG_ERROR, "processNetflowV9: Unsupported Set ID - expected 0/256+, got %d", ntohs(set->id)); } set = (IpfixSetHeader*)((byte*)set + ntohs(set->length)); }  return 0; }   static int processIpfixPacket(IpfixReceiver* ipfixReceiver, byte* message, uint16_t length) { IpfixHeader* header = (IpfixHeader*)message;  if (ntohs(header->length) != length) { DPRINTF("Bad message length - expected %#06x, got %#06x\n", length, ntohs(header->length)); return -1; }   IpfixSetHeader* set = (IpfixSetHeader*)&header->data;   IpfixSetHeader* setX = (IpfixSetHeader*)((char*)message + length);  while (set < setX) { if (ntohs(set->id) == IPFIX_SetId_Template) { processTemplateSet(ipfixReceiver, ntohs(header->sourceId), set); } else if (ntohs(set->id) == IPFIX_SetId_OptionsTemplate) { processOptionsTemplateSet(ipfixReceiver, ntohs(header->sourceId), set); } else if (ntohs(set->id) == IPFIX_SetId_DataTemplate) { processDataTemplateSet(ipfixReceiver, ntohs(header->sourceId), set); } else if (ntohs(set->id) >= IPFIX_SetId_Data_Start) { processDataSet(ipfixReceiver, ntohs(header->sourceId), set); } else { msg(MSG_ERROR, "processIpfixPacket: Unsupported Set ID - expected 2/3/4/256+, got %d", ntohs(set->id)); } set = (IpfixSetHeader*)((byte*)set + ntohs(set->length)); }  return 0; }   static int processMessage(IpfixReceiver* ipfixReceiver, byte* message, uint16_t length) { IpfixHeader* header = (IpfixHeader*)message; if (ntohs(header->version) == 0x000a) { return processIpfixPacket(ipfixReceiver, message, length); } #ifdef SUPPORT_NETFLOWV9 if (ntohs(header->version) == 0x0009) { return processNetflowV9Packet(ipfixReceiver, message, length); } DPRINTF("Bad message version - expected 0x009 or 0x000a, got %#06x\n", ntohs(header->version)); return -1; #else DPRINTF("Bad message version - expected 0x000a, got %#06x\n", ntohs(header->version)); return -1; #endif }  static void printIPv4(FieldType type, FieldData* data) { int octet1 = 0; int octet2 = 0; int octet3 = 0; int octet4 = 0; int imask = 0; if (type.length >= 1) octet1 = data[0]; if (type.length >= 2) octet2 = data[1]; if (type.length >= 3) octet3 = data[2]; if (type.length >= 4) octet4 = data[3]; if (type.length >= 5) imask = data[4]; if (type.length > 5) { DPRINTF("IPv4 Address with length %d unparseable\n", type.length); return; }  if ((type.length == 5) ) { printf("%d.%d.%d.%d/%d", octet1, octet2, octet3, octet4, 32-imask); } else { printf("%d.%d.%d.%d", octet1, octet2, octet3, octet4); } }  static void printPort(FieldType type, FieldData* data) { if (type.length == 0) { printf("zero-length Port"); return; } if (type.length == 2) { int port = ((uint16_t)data[0] << 8)+data[1]; printf("%d", port); return; } if ((type.length >= 4) && ((type.length % 4) == 0)) { int i; for (i = 0; i < type.length; i+=4) { int starti = ((uint16_t)data[i+0] << 8)+data[i+1]; int endi = ((uint16_t)data[i+2] << 8)+data[i+3]; if (i > 0) printf(","); if (starti != endi) { printf("%d:%d", starti, endi); } else { printf("%d", starti); } } return; }  errorf("Port with length %d unparseable", type.length); }  void printProtocol(FieldType type, FieldData* data) { if (type.length != 1) { errorf("Protocol with length %d unparseable", type.length); return; } switch (data[0]) { case IPFIX_protocolIdentifier_ICMP: printf("ICMP"); return; case IPFIX_protocolIdentifier_TCP: printf("TCP"); return; case IPFIX_protocolIdentifier_UDP: printf("UDP"); return; case IPFIX_protocolIdentifier_RAW: printf("RAW"); return; default: printf("unknownProtocol"); return; } }  static void printUint(FieldType type, FieldData* data) { switch (type.length) { case 1: printf("%hhu",*(uint8_t*)data); return; case 2: printf("%hu",ntohs(*(uint16_t*)data)); return; case 4: printf("%u",ntohl(*(uint32_t*)data)); return; case 8: printf("%Lu",ntohll(*(uint64_t*)data)); return; default: errorf("Uint with length %d unparseable", type.length); return; } }   static void* listenerUdpIpv4(void* ipfixReceiver_) { IpfixReceiver* ipfixReceiver = (IpfixReceiver*)ipfixReceiver_;  struct sockaddr_in clientAddress; socklen_t clientAddressLen; byte* data = (byte*)malloc(sizeof(byte)*MAX_MSG_LEN); int n;  while(1) {     clientAddressLen = sizeof(struct sockaddr_in); n = recvfrom(ipfixReceiver->socket, data, MAX_MSG_LEN, 0, (struct sockaddr*)&clientAddress, &clientAddressLen);  if(n < 0) { msg(MSG_DEBUG, "recvfrom returned with error, terminating UDP/IPv4 listener thread"); break; }  pthread_mutex_lock(&ipfixReceiver->mutex); processMessage(ipfixReceiver, data, n); pthread_mutex_unlock(&ipfixReceiver->mutex); }  free(data);  return 0; }     int initializeIpfixReceivers() { return 0; }   int deinitializeIpfixReceivers() { return 0; }   void printFieldData(FieldType type, FieldData* pattern) { char* s;  switch (type.id) { case IPFIX_TYPEID_protocolIdentifier: printf("protocolIdentifier:"); printProtocol(type, pattern); break; case IPFIX_TYPEID_sourceIPv4Address: printf("sourceIPv4Address:"); printIPv4(type, pattern); break; case IPFIX_TYPEID_destinationIPv4Address: printf("destinationIPv4Address:"); printIPv4(type, pattern); break; case IPFIX_TYPEID_sourceTransportPort: printf("sourceTransportPort:"); printPort(type, pattern); break; case IPFIX_TYPEID_destinationtransportPort: printf("destinationtransportPort:"); printPort(type, pattern); break; default: s = typeid2string(type.id); if (s != NULL) { printf("%s:", s); printUint(type, pattern); } else { DPRINTF("Field with ID %d unparseable\n", type.id); } break; } }   FieldInfo* getTemplateFieldInfo(TemplateInfo* ti, FieldType* type) { int i;  for (i = 0; i < ti->fieldCount; i++) { if ((ti->fieldInfo[i].type.id == type->id) && (ti->fieldInfo[i].type.eid == type->eid)) { return &ti->fieldInfo[i]; } }  return NULL; }   FieldInfo* getDataTemplateFieldInfo(DataTemplateInfo* ti, FieldType* type) { int i;  for (i = 0; i < ti->fieldCount; i++) { if ((ti->fieldInfo[i].type.id == type->id) && (ti->fieldInfo[i].type.eid == type->eid)) { return &ti->fieldInfo[i]; } }  return NULL; }   FieldInfo* getDataTemplateDataInfo(DataTemplateInfo* ti, FieldType* type) { int i;  for (i = 0; i < ti->dataCount; i++) { if ((ti->dataInfo[i].type.id == type->id) && (ti->dataInfo[i].type.eid == type->eid)) { return &ti->dataInfo[i]; } }  return NULL; }   void addIpfixReceiverCallbacks(IpfixReceiver* ipfixReceiver, CallbackInfo handles) { int n = ++ipfixReceiver->callbackCount; ipfixReceiver->callbackInfo = (CallbackInfo*)realloc(ipfixReceiver->callbackInfo, n * sizeof(CallbackInfo)); memcpy(&ipfixReceiver->callbackInfo[n-1], &handles, sizeof(CallbackInfo)); }   IpfixReceiver* createIpfixReceiver(uint16_t port) { IpfixReceiver* ipfixReceiver; struct sockaddr_in serverAddress;  if(!(ipfixReceiver=(IpfixReceiver*)malloc(sizeof(IpfixReceiver)))) { msg(MSG_FATAL, "Ran out of memory"); goto out0; }  ipfixReceiver->callbackInfo = 0;  if(!(ipfixReceiver->templateBuffer = createTemplateBuffer(ipfixReceiver))) { msg(MSG_FATAL, "Could not create template Buffer"); goto out1; }  if (pthread_mutex_init(&ipfixReceiver->mutex, NULL) != 0) { msg(MSG_FATAL, "Could not init mutex"); goto out2; }  if (pthread_mutex_lock(&ipfixReceiver->mutex) != 0) { msg(MSG_FATAL, "Could not lock mutex"); goto out2; }  ipfixReceiver->socket = socket(AF_INET, SOCK_DGRAM, 0); if(ipfixReceiver->socket < 0) { msg(MSG_FATAL, "Could not create socket"); goto out2; }  serverAddress.sin_family = AF_INET; serverAddress.sin_addr.s_addr = htonl(INADDR_ANY); serverAddress.sin_port = htons(port); if(bind(ipfixReceiver->socket, (struct sockaddr*)&serverAddress, sizeof(struct sockaddr_in)) < 0) { msg(MSG_FATAL, "Could not bind socket"); goto out3; }  if(pthread_create(&(ipfixReceiver->thread), 0, listenerUdpIpv4, ipfixReceiver) != 0) { msg(MSG_FATAL, "Could not pthread_create listener thread"); goto out3; }   return ipfixReceiver;  out3: close(ipfixReceiver->socket); out2: destroyTemplateBuffer(ipfixReceiver->templateBuffer); out1: free(ipfixReceiver); out0: return NULL; }   int startIpfixReceiver(IpfixReceiver* ipfixReceiver) { if (pthread_mutex_unlock(&ipfixReceiver->mutex) != 0) { return -1; }  return 0; }   int stopIpfixReceiver(IpfixReceiver* ipfixReceiver) { if (pthread_mutex_lock(&ipfixReceiver->mutex) != 0) { return -1; }  return 0; }   void destroyIpfixReceiver(IpfixReceiver* ipfixReceiver) { close(ipfixReceiver->socket);  if (pthread_mutex_unlock(&ipfixReceiver->mutex) != 0) { DPRINTF("Could not unlock mutex\n"); }  destroyTemplateBuffer(ipfixReceiver->templateBuffer); free(ipfixReceiver->callbackInfo); free(ipfixReceiver); }
static void processDataSet(IpfixReceiver* ipfixReceiver, SourceID sourceId, IpfixSetHeader* set) {	BufferedTemplate* bt = getBufferedTemplate(ipfixReceiver->templateBuffer, sourceId, ntohs(set->id));	if(bt == 0) {                /* this error may come in rapid succession; I hope I don't regret it */		msg(MSG_INFO, "Template %d unknown to collecting process", ntohs(set->id));		return;	}		#ifdef SUPPORT_NETFLOWV9	if ((bt->setID == IPFIX_SetId_Template) || (bt->setID == NetflowV9_SetId_Template)) {	#else	if (bt->setID == IPFIX_SetId_Template) {	#endif		TemplateInfo* ti = bt->templateInfo;		uint16_t length = ntohs(set->length)-((byte*)(&set->data)-(byte*)set);		byte* record = &set->data;			if (bt->recordLength < 65535) {			byte* recordX = record+length;			if (record >= recordX - (bt->recordLength - 1)) {				DPRINTF("Got a Data Set that contained not a single full record\n");			}			/* We stop processing when no full record is left */			while (record < recordX - (bt->recordLength - 1)) {				int n;						for (n = 0; n < ipfixReceiver->callbackCount; n++) {					CallbackInfo* ci = &ipfixReceiver->callbackInfo[n];					if (ci->dataRecordCallbackFunction) ci->dataRecordCallbackFunction(ci->handle, sourceId, ti, bt->recordLength, record);					}				record = record + bt->recordLength;				}			} else {			byte* recordX = record+length;			if (record >= recordX - 3) {				DPRINTF("Got a Data Set that contained not a single full record\");			}			/* We assume that all variable-length records are >= 4 byte, so we stop processing when only 3 bytes are left */			while (record < recordX - 3) {				int recordLength=0;				int i;				for (i = 0; i < ti->fieldCount; i++) {					int fieldLength = 0;					if (ti->fieldInfo[i].type.length < 65535) {						fieldLength = ti->fieldInfo[i].type.length;						} else {						if (*(byte*)record < 255) {							fieldLength = *(byte*)record;							} else {							fieldLength = *(uint16_t*)(record+1);							}						}					ti->fieldInfo[i].type.length = fieldLength;					ti->fieldInfo[i].offset = recordLength;					recordLength += fieldLength;					}				int n;						for (n = 0; n < ipfixReceiver->callbackCount; n++) {					CallbackInfo* ci = &ipfixReceiver->callbackInfo[n];					if (ci->dataRecordCallbackFunction) ci->dataRecordCallbackFunction(ci->handle, sourceId, ti, recordLength, record);					}				record = record + recordLength;				}			}		} else 	if (bt->setID == IPFIX_SetId_OptionsTemplate) {  			OptionsTemplateInfo* ti = bt->optionsTemplateInfo;		uint16_t length = ntohs(set->length)-((byte*)(&set->data)-(byte*)set);		byte* record = &set->data;			if (bt->recordLength < 65535) {			byte* recordX = record+length;			while (record < recordX) {				int n;						for (n = 0; n < ipfixReceiver->callbackCount; n++) {					CallbackInfo* ci = &ipfixReceiver->callbackInfo[n];					if (ci->optionsRecordCallbackFunction) ci->optionsRecordCallbackFunction(ci->handle, sourceId, ti, bt->recordLength, record);					}				record = record + bt->recordLength;				}			} else {			byte* recordX = record+length;			while (record < recordX) {				int recordLength=0;				int i;				for (i = 0; i < ti->scopeCount; i++) {					int fieldLength = 0;					if (ti->scopeInfo[i].type.length < 65535) {						fieldLength = ti->scopeInfo[i].type.length;						} else {						if (*(byte*)record < 255) {							fieldLength = *(byte*)record;							} else {							fieldLength = *(uint16_t*)(record+1);							}						}					ti->scopeInfo[i].type.length = fieldLength;					ti->scopeInfo[i].offset = recordLength;					recordLength += fieldLength;					}				for (i = 0; i < ti->fieldCount; i++) {					int fieldLength = 0;					if (ti->fieldInfo[i].type.length < 65535) {						fieldLength = ti->fieldInfo[i].type.length;						} else {						if (*(byte*)record < 255) {							fieldLength = *(byte*)record;							} else {							fieldLength = *(uint16_t*)(record+1);							}						}					ti->fieldInfo[i].type.length = fieldLength;					ti->fieldInfo[i].offset = recordLength;					recordLength += fieldLength;					}				int n;						for (n = 0; n < ipfixReceiver->callbackCount; n++) {					CallbackInfo* ci = &ipfixReceiver->callbackInfo[n];					if (ci->optionsRecordCallbackFunction) ci->optionsRecordCallbackFunction(ci->handle, sourceId, ti, recordLength, record);					}				record = record + recordLength;				}			}		} else   	if (bt->setID == IPFIX_SetId_DataTemplate) {		DataTemplateInfo* ti = bt->dataTemplateInfo;		uint16_t length = ntohs(set->length)-((byte*)(&set->data)-(byte*)set);		byte* record = &set->data;		if (bt->recordLength < 65535) {			byte* recordX = record+length;			while (record < recordX) {				int n;						for (n = 0; n < ipfixReceiver->callbackCount; n++) {					CallbackInfo* ci = &ipfixReceiver->callbackInfo[n];					if (ci->dataDataRecordCallbackFunction) ci->dataDataRecordCallbackFunction(ci->handle, sourceId, ti, bt->recordLength, record);					}				record = record + bt->recordLength;				}			} else {			byte* recordX = record+length;			while (record < recordX) {				int recordLength=0;				int i;				for (i = 0; i < ti->fieldCount; i++) {					int fieldLength = 0;					if (ti->fieldInfo[i].type.length < 65535) {						fieldLength = ti->fieldInfo[i].type.length;						} else {						if (*(byte*)record < 255) {							fieldLength = *(byte*)record;							} else {							fieldLength = *(uint16_t*)(record+1);							}						}					ti->fieldInfo[i].type.length = fieldLength;					ti->fieldInfo[i].offset = recordLength;					recordLength += fieldLength;					}				int n;						for (n = 0; n < ipfixReceiver->callbackCount; n++) {					CallbackInfo* ci = &ipfixReceiver->callbackInfo[n];					if (ci->dataDataRecordCallbackFunction) ci->dataDataRecordCallbackFunction(ci->handle, sourceId, ti, recordLength, record);					}				record = record + recordLength;				}			}		} else {		msg(MSG_FATAL, "Data Set based on known but unhandled template type %d", bt->setID);		}}	/** * Process a NetflowV9 Packet * @return 0 on success */	static int processNetflowV9Packet(IpfixReceiver* ipfixReceiver, byte* message, uint16_t length) {	NetflowV9Header* header = (NetflowV9Header*)message;		/* pointer to first set */	IpfixSetHeader* set = (IpfixSetHeader*)&header->data;	int i;	for (i = 0; i < ntohs(header->setCount); i++) {		if (ntohs(set->id) == NetflowV9_SetId_Template) {  			processTemplateSet(ipfixReceiver, ntohs(header->sourceId), set);  			} else		if (ntohs(set->id) >= IPFIX_SetId_Data_Start) {  			processDataSet(ipfixReceiver, ntohs(header->sourceId), set);  			} else {			msg(MSG_ERROR, "processNetflowV9: Unsupported Set ID - expected 0/256+, got %d", ntohs(set->id));  			}		set = (IpfixSetHeader*)((byte*)set + ntohs(set->length));		}	return 0;}/** * Process an IPFIX Packet * @return 0 on success */	static int processIpfixPacket(IpfixReceiver* ipfixReceiver, byte* message, uint16_t length) {	IpfixHeader* header = (IpfixHeader*)message;	if (ntohs(header->length) != length) { 		DPRINTF("Bad message length - expected %#06x, got %#06x\n", length, ntohs(header->length));		return -1;	}	/* pointer to first set */	IpfixSetHeader* set = (IpfixSetHeader*)&header->data;	/* pointer beyond message */	IpfixSetHeader* setX = (IpfixSetHeader*)((char*)message + length); 	while (set < setX) {		if (ntohs(set->id) == IPFIX_SetId_Template) {  			processTemplateSet(ipfixReceiver, ntohs(header->sourceId), set);		} else if (ntohs(set->id) == IPFIX_SetId_OptionsTemplate) {  			processOptionsTemplateSet(ipfixReceiver, ntohs(header->sourceId), set);		} else if (ntohs(set->id) == IPFIX_SetId_DataTemplate) {  			processDataTemplateSet(ipfixReceiver, ntohs(header->sourceId), set);		} else if (ntohs(set->id) >= IPFIX_SetId_Data_Start) {  			processDataSet(ipfixReceiver, ntohs(header->sourceId), set);		} else {			msg(MSG_ERROR, "processIpfixPacket: Unsupported Set ID - expected 2/3/4/256+, got %d", ntohs(set->id));		}		set = (IpfixSetHeader*)((byte*)set + ntohs(set->length));	}	return 0;}/** * Process new Message * @return 0 on success */	static int processMessage(IpfixReceiver* ipfixReceiver, byte* message, uint16_t length) {	IpfixHeader* header = (IpfixHeader*)message;	if (ntohs(header->version) == 0x000a) {		return processIpfixPacket(ipfixReceiver, message, length);		}	#ifdef SUPPORT_NETFLOWV9	if (ntohs(header->version) == 0x0009) {		return processNetflowV9Packet(ipfixReceiver, message, length);		}	DPRINTF("Bad message version - expected 0x009 or 0x000a, got %#06x\n", ntohs(header->version));	return -1;	#else	DPRINTF("Bad message version - expected 0x000a, got %#06x\n", ntohs(header->version));	return -1;	#endif	}	static void printIPv4(FieldType type, FieldData* data) {	int octet1 = 0;	int octet2 = 0;	int octet3 = 0;	int octet4 = 0;	int imask = 0;	if (type.length >= 1) octet1 = data[0];	if (type.length >= 2) octet2 = data[1];	if (type.length >= 3) octet3 = data[2];	if (type.length >= 4) octet4 = data[3];	if (type.length >= 5) imask = data[4];	if (type.length > 5) {		DPRINTF("IPv4 Address with length %d unparseable\n", type.length);		return;		}		if ((type.length == 5) /*&& (imask != 0)*/) {		printf("%d.%d.%d.%d/%d", octet1, octet2, octet3, octet4, 32-imask);		} else {		printf("%d.%d.%d.%d", octet1, octet2, octet3, octet4);		}	}static void printPort(FieldType type, FieldData* data) {	if (type.length == 0) {		printf("zero-length Port");		return;		}	if (type.length == 2) {		int port = ((uint16_t)data[0] << 8)+data[1];		printf("%d", port);		return;		}	if ((type.length >= 4) && ((type.length % 4) == 0)) {		int i;		for (i = 0; i < type.length; i+=4) {			int starti = ((uint16_t)data[i+0] << 8)+data[i+1];			int endi = ((uint16_t)data[i+2] << 8)+data[i+3];			if (i > 0) printf(",");			if (starti != endi) {				printf("%d:%d", starti, endi);				} else {				printf("%d", starti);				}			} 		return;		}		errorf("Port with length %d unparseable", type.length);	}void printProtocol(FieldType type, FieldData* data) {	if (type.length != 1) {		errorf("Protocol with length %d unparseable", type.length);		return;		}	switch (data[0]) {		case IPFIX_protocolIdentifier_ICMP:			printf("ICMP");			return;		case IPFIX_protocolIdentifier_TCP:			printf("TCP");			return;		case IPFIX_protocolIdentifier_UDP: 			printf("UDP");			return;		case IPFIX_protocolIdentifier_RAW: 			printf("RAW");			return;		default:			printf("unknownProtocol");			return;		}	}static void printUint(FieldType type, FieldData* data) {	switch (type.length) {		case 1:			printf("%hhu",*(uint8_t*)data);			return;		case 2:			printf("%hu",ntohs(*(uint16_t*)data));			return;		case 4:			printf("%u",ntohl(*(uint32_t*)data));			return;		case 8:			printf("%Lu",ntohll(*(uint64_t*)data));			return;		default:			errorf("Uint with length %d unparseable", type.length);			return;		}	}/* FIXME: implement clean exiting Use pthread_sigmask() ? */static void* listenerUdpIpv4(void* ipfixReceiver_) {	IpfixReceiver* ipfixReceiver = (IpfixReceiver*)ipfixReceiver_;		struct sockaddr_in clientAddress;	socklen_t clientAddressLen;	byte* data = (byte*)malloc(sizeof(byte)*MAX_MSG_LEN);	int n;		while(1) {			//static uint16_t packets = 0;		//if (packets++ >= 10000) break;				clientAddressLen = sizeof(struct sockaddr_in);		n = recvfrom(ipfixReceiver->socket, data, MAX_MSG_LEN, 0, (struct sockaddr*)&clientAddress, &clientAddressLen);		if(n < 0) {			msg(MSG_DEBUG, "recvfrom returned with error, terminating UDP/IPv4 listener thread");			break;		}      		pthread_mutex_lock(&ipfixReceiver->mutex);		processMessage(ipfixReceiver, data, n);		pthread_mutex_unlock(&ipfixReceiver->mutex);	}	free(data);	return 0;}/***** Exported Functions ****************************************************//** * Initializes internal data. * Call once before using any function in this module * @return 0 if call succeeded */int initializeIpfixReceivers() {	return 0;}/** * Destroys internal data. * Call once to tidy up. Do not use any function in this module afterwards * @return 0 if call succeeded */int deinitializeIpfixReceivers() {	return 0;}/** * Prints a string representation of FieldData to stdout. */void printFieldData(FieldType type, FieldData* pattern) {	char* s;		switch (type.id) {		case IPFIX_TYPEID_protocolIdentifier:			printf("protocolIdentifier:");			printProtocol(type, pattern);			break;		case IPFIX_TYPEID_sourceIPv4Address:			printf("sourceIPv4Address:");			printIPv4(type, pattern);			break;		case IPFIX_TYPEID_destinationIPv4Address:			printf("destinationIPv4Address:");			printIPv4(type, pattern);			break;						case IPFIX_TYPEID_sourceTransportPort:			printf("sourceTransportPort:");			printPort(type, pattern);			break;		case IPFIX_TYPEID_destinationtransportPort:			printf("destinationtransportPort:");			printPort(type, pattern);			break;		default:			s = typeid2string(type.id);			if (s != NULL) {				printf("%s:", s);				printUint(type, pattern);				} else {				DPRINTF("Field with ID %d unparseable\n", type.id);				}			break;		}}/** * Gets a Template's FieldInfo by field id. * @param ti Template to search in * @param type Field id and field eid to look for, length is ignored * @return NULL if not found */FieldInfo* getTemplateFieldInfo(TemplateInfo* ti, FieldType* type) {	int i;		for (i = 0; i < ti->fieldCount; i++) {		if ((ti->fieldInfo[i].type.id == type->id) && (ti->fieldInfo[i].type.eid == type->eid)) {			return &ti->fieldInfo[i];			}		}	return NULL;}/** * Gets a DataTemplate's FieldInfo by field id. * @param ti DataTemplate to search in * @param type Field id and field eid to look for, length is ignored * @return NULL if not found */FieldInfo* getDataTemplateFieldInfo(DataTemplateInfo* ti, FieldType* type) {	int i;		for (i = 0; i < ti->fieldCount; i++) {		if ((ti->fieldInfo[i].type.id == type->id) && (ti->fieldInfo[i].type.eid == type->eid)) {			return &ti->fieldInfo[i];			}		}	return NULL;}/** * Gets a DataTemplate's Data-FieldInfo by field id. * @param ti DataTemplate to search in * @param type Field id and field eid to look for, length is ignored * @return NULL if not found */FieldInfo* getDataTemplateDataInfo(DataTemplateInfo* ti, FieldType* type) {	int i;		for (i = 0; i < ti->dataCount; i++) {		if ((ti->dataInfo[i].type.id == type->id) && (ti->dataInfo[i].type.eid == type->eid)) {			return &ti->dataInfo[i];			}		}	return NULL;		}	/** * Adds a set of callback functions to the list of functions to call when a new Message arrives * @param ipfixReceiver IpfixReceiver to set the callback function for * @param handles set of callback functions */void addIpfixReceiverCallbacks(IpfixReceiver* ipfixReceiver, CallbackInfo handles) {	int n = ++ipfixReceiver->callbackCount;	ipfixReceiver->callbackInfo = (CallbackInfo*)realloc(ipfixReceiver->callbackInfo, n * sizeof(CallbackInfo));	memcpy(&ipfixReceiver->callbackInfo[n-1], &handles, sizeof(CallbackInfo));}/** * Creates a new IpfixReceiver. * Call @c startIpfixReceiver() to start processing messages. * @param port Port to listen on * @return handle for further interaction */IpfixReceiver* createIpfixReceiver(uint16_t port) {	IpfixReceiver* ipfixReceiver;	struct sockaddr_in serverAddress;		if(!(ipfixReceiver=(IpfixReceiver*)malloc(sizeof(IpfixReceiver)))) {		msg(MSG_FATAL, "Ran out of memory");		goto out0;	}	ipfixReceiver->callbackInfo = 0;		if(!(ipfixReceiver->templateBuffer = createTemplateBuffer(ipfixReceiver))) {		msg(MSG_FATAL, "Could not create template Buffer");		goto out1;	}	if (pthread_mutex_init(&ipfixReceiver->mutex, NULL) != 0) {		msg(MSG_FATAL, "Could not init mutex");		goto out2;	}			if (pthread_mutex_lock(&ipfixReceiver->mutex) != 0) {		msg(MSG_FATAL, "Could not lock mutex");		goto out2;	}	ipfixReceiver->socket = socket(AF_INET, SOCK_DGRAM, 0);	if(ipfixReceiver->socket < 0) {		msg(MSG_FATAL, "Could not create socket");		goto out2;	}	serverAddress.sin_family = AF_INET;	serverAddress.sin_addr.s_addr = htonl(INADDR_ANY);	serverAddress.sin_port = htons(port);	if(bind(ipfixReceiver->socket, (struct sockaddr*)&serverAddress, sizeof(struct sockaddr_in)) < 0) {		msg(MSG_FATAL, "Could not bind socket");		goto out3;	}	if(pthread_create(&(ipfixReceiver->thread), 0, listenerUdpIpv4, ipfixReceiver) != 0) {		msg(MSG_FATAL, "Could not pthread_create listener thread");		goto out3;	}	//listenerUdpIpv4(ipfixReceiver); //debug - single-threaded		return ipfixReceiver;out3:	close(ipfixReceiver->socket);out2:	destroyTemplateBuffer(ipfixReceiver->templateBuffer);out1:	free(ipfixReceiver);out0:	return NULL;}/** * Starts processing messages. * All sockets prepared by calls to createIpfixReceiver() will start * receiving messages until stopIpfixReceiver() is called. * @return 0 on success, non-zero on error */int startIpfixReceiver(IpfixReceiver* ipfixReceiver) {	if (pthread_mutex_unlock(&ipfixReceiver->mutex) != 0) {		return -1;	}		return 0;}	/** * Stops processing messages. * No more messages will be processed until the next startIpfixReceiver() call. * @return 0 on success, non-zero on error */int stopIpfixReceiver(IpfixReceiver* ipfixReceiver) {	if (pthread_mutex_lock(&ipfixReceiver->mutex) != 0) {		return -1;	}	return 0;}/** * Frees memory used by a IpfixReceiver. * @param ipfixReceiver Handle returned by @c createIpfixReceiver() */void destroyIpfixReceiver(IpfixReceiver* ipfixReceiver) {	close(ipfixReceiver->socket);		if (pthread_mutex_unlock(&ipfixReceiver->mutex) != 0) {		DPRINTF("Could not unlock mutex\n");	}		destroyTemplateBuffer(ipfixReceiver->templateBuffer);	free(ipfixReceiver->callbackInfo);	free(ipfixReceiver);}