newtella.c

linux下支持P2P的客户端程序,采用了Gnutella的方法,其中有实现Gnutella的具体源码,是一个基于p2p环境下的音乐共享软件,有助于对P2P的研究

	rmsg = newtella_malloc(sizeof *rmsg);

	rmsg->head = newtella_malloc(sizeof *(rmsg->head));
	*(rmsg->head) = *gh;

	rmsg->length = GN_HEADER_SIZE + gh->length;
	rmsg->policy = RP_ALL_EXC;
	rmsg->con    = con;
	rmsg->time   = time((time_t *)NULL);

	/* route msg */
	route_msg_add(rmsg, buf);

 skip_query:	
	newtella_free(gquery->name); 
 	newtella_free(gquery);

	return gh->length;
}

int handle_search_response(struct newtella_connection *con, struct gnutella_header *gh, char *buf)
{
	struct gnutella_hits *ghits;
	struct gnutella_hit  *ghit;
	struct file_found    *found;
	struct squery        *query;
	struct message       *msg;
	struct message       *rmsg;

	gint    i;
	gint    last_found      = gl_options->files_found;
	gint    forus           = 0;
	gchar   guid_for_push[GUID_LEN];
	gchar   vendor_id[4]; 
	gchar   open_data_size;
	gchar   open_data_byte[2];
	GSList *l;

	/* GTK stuff */
	gchar *dummy[6] = { "", "", "", "", "", "" };

	/* check if we have seen the response */
	if (route_msg_find(gh->guid, GNUTELLA_SEARCHRESPONSE, gh->length)) {  
		con->total_received_dropped++;
		return gh->length;
	} else
		con->total_handled++;
   
	/* is it for us? */
	for (l = gl_queries; l; l = l->next) {
		query = (struct squery *) l->data;
		if (guidcmp(query->guid, gh->guid))
			forus = 1;
	}

	if (!forus)
		goto notforus;

    	/* store the guid to use it for push */
	//guid_for_push = newtella_malloc(GUID_LEN);
	memcpy(&guid_for_push, buf+gh->length-16, 16);

	ghits = newtella_malloc(sizeof *ghits);
	GNUTELLA_HITS_GRAB(ghits, buf)

	buf += GN_HITS_SIZE;
	ghit = newtella_malloc(sizeof *ghit);

	for (i=0; i<ghits->n; i++) {

		/* check if we reached the maximum results limit */
		if (gl_options->max_search_results == gl_options->files_found)
			goto skip;

		GNUTELLA_HIT_GRAB(ghit, buf)

		/* check if we really have mp3 results */
			if (file_is_mp3(ghit->name)) {
			found = newtella_malloc(sizeof(struct file_found));
			found->file_name = newtella_malloc(strlen(ghit->name)+1);
			strcpy(found->file_name, ghit->name);

			guidcpy(found->guid, guid_for_push);
			found->index = ghit->index;
			found->size  = ghit->size;
			found->port  = ghits->port;
			found->ip    = g_htonl(ghits->ip);
			found->speed = ghits->speed;

			if (host_is_private(found->ip, found->port))
				found->push = 1;
			else
				found->push = 0;
			
			gl_results = g_list_append(gl_results, found);
			gl_options->files_found++;

			gtk_clist_append(GTK_CLIST(search_list), dummy);
		
		}   
		
		/* go to next hit */
		buf += GN_HIT_SIZE + strlen(ghit->name);
		while (memcmp(buf++, "\0", 1));

		/* data between the 2 '\n's */
		while (memcmp(buf++, "\0", 1));
		
		newtella_free(ghit->name);
	}

	/* no trailer */
	if (guidcmp(buf, guid_for_push))
		goto skip;

	/* parse the Trailer of the extended QHD */
	memcpy(&vendor_id, buf, 4);
	vendor_id[4] = '\0';

	buf += 4;

	memcpy(&open_data_size, buf, 1);
	if (open_data_size == 2) {
		memcpy(&open_data_byte[0], buf+1, 1);
		memcpy(&open_data_byte[1], buf+2, 1);

		rate_result_set(last_found, open_data_byte, vendor_id);
	}	

 skip:
	newtella_free(ghit);
	newtella_free(ghits);

	gui_update_found_results(last_found, translate_vendor_id(vendor_id));
	
	return gh->length;
 notforus:

        /* remove packet from network if it has expired */
	if (gh->ttl == 1) 
		return gh->length;

	/* check if we haven't seen the search related to the result set */
        msg = route_msg_find(gh->guid, GNUTELLA_SEARCH, 0);

	if (msg == NULL || msg->con == NULL)
		return gh->length;

        /* ok, we have to route it */
	rmsg = newtella_malloc(sizeof *rmsg);

	rmsg->head = newtella_malloc(sizeof *(rmsg->head));
	*(rmsg->head) = *gh;

	guidcpy(msg->guid, buf+gh->length-16);
	rmsg->length = GN_HEADER_SIZE + gh->length;
	rmsg->policy = RP_ONLY_TO;
	rmsg->con    = msg->con;
	rmsg->time   = time((time_t *)NULL);

	/* route msg */
	route_msg_add(rmsg, buf);
	 	
	return gh->length;
}

int handle_packet(struct newtella_connection *con, struct gnutella_header *gh, char *buf)
{
	switch (gh->f) {
	case GNUTELLA_INIT:
		return handle_init(con, gh);
		break;
	case GNUTELLA_INITRESPONSE:
		return handle_init_response(con, buf, gh);
		break;
	case GNUTELLA_PUSHREQUEST:
		return handle_push(con, gh, buf);
		break;
	case GNUTELLA_SEARCH:
		return handle_search(con, gh, buf);
		break;
	case GNUTELLA_SEARCHRESPONSE:
		return handle_search_response(con, gh, buf);
		break;
	default:
		break;
	}
	
	return 0;
}

void newtella_connect(gpointer gdata, gint s, GdkInputCondition cond)
{
	struct newtella_connection *con = (struct newtella_connection *)gdata;
	gint r, option, size;

	g_return_if_fail(con);

	if (con->write_tag)
		gdk_input_remove(con->write_tag);

	if (cond == GDK_INPUT_EXCEPTION) {
		con_mark(con, GS_FAILED);
		return;
	}

	r = getsockopt(con->s, SOL_SOCKET, SO_ERROR, (void *) &option, &size);

	if (r == -1 || option) {
		/* failed to connect */
		con_mark(con, GS_FAILED);
		return;
	} else {
		con->state = GS_HANDSHAKE;
		if (con->type != CT_UPLOAD)
			gui_update_con_status(con);

		con->write_tag = gdk_input_add(con->s, 
					       GDK_INPUT_WRITE | GDK_INPUT_EXCEPTION, 
					       (GdkInputFunction) newtella_write, 
					       (gpointer)con);

		/* try to guess our ip */
		if (host_is_private(gl_my_host_info->ip, gl_my_host_info->port))
			newtella_guess_ip(con->s);
	}

}

void newtella_write(gpointer gdata, gint s, GdkInputCondition cond)
{
	struct gpacket *packet;
	struct newtella_connection *con = (struct newtella_connection *)gdata;
	gint r;
	
	g_return_if_fail(con);

	if (con->write_tag)
		gdk_input_remove(con->write_tag);
	
	if (cond == GDK_INPUT_EXCEPTION) {
		con_mark(con, GS_FAILED);
		return;
	}
	
	switch (con->state) {
	case GS_HANDSHAKE:
		if (con->type == CT_OUTGOING) {
			
			con->write_tag = 0;
			
                        /* send outgoing handshake */
			r = send(con->s, "GNUTELLA CONNECT/0.4\n\n", 
				 strlen("GNUTELLA CONNECT/0.4\n\n"), 0);

			if (r < 0) 
				con_mark(con, GS_FAILED);
			      
		}
		if (con->type == CT_UPLOAD) {
			
			packet = (struct gpacket *) g_list_last(con->queue)->data;
			
			r = send(con->s, packet->data, packet->len, 0);
						
			if (r == packet->len) {
				
				con->read_tag = gdk_input_add(con->s, 
							      GDK_INPUT_READ | GDK_INPUT_EXCEPTION, 
							      (GdkInputFunction) newtella_read, 
							      (gpointer)con);

			} else 
				con_mark(con, GS_FAILED);
			
			
			con->queue = g_list_remove(con->queue, packet);
			newtella_free(packet->data);
			newtella_free(packet);
		}
		break;
	case GS_HOSTFEED:
	case GS_CONNECTED:
		while (g_list_length(con->queue)) {
			packet = (struct gpacket *) g_list_last(con->queue)->data;
			
			r = send(con->s, packet->data, packet->len, 0);
			
			if (r > 0) 
				con->total_sent++;
			else {
				con->total_dropped_to_send++;
				break;
			}

			if (r < packet->len && r > 0) {
				memcpy(packet->data, packet->data+r, packet->len-r);
				packet->len -= r;
				break;
			}
			else
				con->queue = g_list_remove(con->queue, packet);
			
			newtella_free(packet->data);
			newtella_free(packet);
			
			con->last_time = time((time_t)NULL);
		};
			
		gui_update_con_status(con);
	    
		con->write_tag = 0;
	  
		break;
	case GS_FAILED:
		break;
	case GS_TIMEOUT:
		break;
	default:
		break;
	}

}

void newtella_read(gpointer gdata, gint s, GdkInputCondition cond)
{
	struct newtella_connection *con = (struct newtella_connection *)gdata;
	struct gnutella_header *gh;
	guint16 bytes_readed = 0, bytes_to_read = 0, 
		bytes_parsed = 0, bytes_left = 0,
		bytes_done = 0;
	char *p;

	g_return_if_fail(con);

	if (cond == GDK_INPUT_EXCEPTION) {
		con_mark(con, GS_FAILED);
		gdk_input_remove(con->read_tag);
		return;
	}
	
	bytes_to_read = bytes_avail(con->s);
	
	if (!bytes_to_read || bytes_to_read > MAX_PACKET_LENGTH) {
		con_mark(con, GS_FAILED);
		gdk_input_remove(con->read_tag);
		return;
	}
	
	con->b = con->buffer + con->buffer_len;
	if (((con->b-con->buffer)+bytes_to_read) > MAX_PACKET_LENGTH) {
		/* FIXME */
		con_mark(con, GS_FAILED);
		gdk_input_remove(con->read_tag);
		return;
	}

	con->buffer_len += bytes_to_read;
	bytes_readed = con_recv_packet(con, bytes_to_read);
       
	if (bytes_readed != bytes_to_read) {
		con_mark(con, GS_FAILED);
		gdk_input_remove(con->read_tag);
		return;
	}

	con->last_time = time((time_t) NULL);

	p = con->buffer;
	gh = newtella_malloc(sizeof *gh);
     
	do {
		if ((con->buffer_len - bytes_parsed) < 0)
			g_print("CRASH again: %d\n", bytes_left);
 
		bytes_left = con->buffer_len - bytes_parsed;
	 
		switch (con->state) {
		case GS_FAILED:
			gdk_input_remove(con->read_tag);
			newtella_free(gh);
			return;
		case GS_TIMEOUT:
			gdk_input_remove(con->read_tag);
			newtella_free(gh);
			return;
		case GS_HOSTFEED:
			newtella_free(gh);
			if (g_list_length(con->queue) == 0)
				con_mark(con, GS_FAILED);
			return;
		case GS_CONNECTED:						
			if (bytes_left >= GN_HEADER_SIZE)
				GNUTELLA_HDR_GRAB(gh, p)
			else {	
				memcpy(con->buffer, p, bytes_left);
				con->buffer_len = bytes_left;
				goto skip;
			}
		       
			if (gh->length > (bytes_left-GN_HEADER_SIZE)) {
				memcpy(con->buffer, p, bytes_left);
				con->buffer_len = bytes_left;
				goto skip;
			}

			if (gh->length > (MAX_PACKET_LENGTH  - con->buffer_len)) { 
				g_print("DANGER! plen: %d buflen: %d\n", 
					gh->length, con->buffer_len);
				
				con_mark(con, GS_FAILED);
				gdk_input_remove(con->read_tag);
				return;
			}

			p += GN_HEADER_SIZE;
			bytes_parsed += GN_HEADER_SIZE;
			
			if (newtella_packet_sucks(gh)) {
				con->total_bad_received++;
				if (gh->length + (p - con->buffer) < MAX_PACKET_LENGTH)
					bytes_done += gh->length;
				else {
					con_mark(con, GS_FAILED);
					gdk_input_remove(con->read_tag);
					return;
				}
			}			
			
			if (bytes_left > gl_options->con_speed*100 &&
			    (gh->f == GNUTELLA_INIT || gh->f == GNUTELLA_SEARCH)) {
				con->total_received_dropped++;
				bytes_done += gh->length;
			} else if (bytes_left > MAX_BYTES_PER_SECOND*10) {
				con->total_received_dropped++;
				bytes_done += gh->length;
			} else
				bytes_done = handle_packet(con, gh, p);
			
			p += bytes_done;
			bytes_parsed += bytes_done;

			break;
		case GS_HANDSHAKE:
			if (newtella_handshake_sucks(p, bytes_left)) {
				con_mark(con, GS_FAILED);
				return;
			}
			/* outgoing connection waiting to establish */
			if (handle_connect_response(con, con->buffer)) { 
				p += strlen("GNUTELLA OK\n\n");
				bytes_parsed += strlen("GNUTELLA OK\n\n");
				break;
			} /* incoming connection waiting to establish */
			else if (handle_connect(con, con->buffer)) {
				p += strlen("GNUTELLA CONNECT/0.4\n\n");
				bytes_parsed += strlen("GNUTELLA CONNECT/0.4\n\n");
				break;
			} /* download connection waiting to establish */
			else if (handle_get_request(con, con->buffer)) {
				upload_start(con, p);
				return;
			}  
	
			bytes_parsed = con->buffer_len;
			break;
		default:
			break;
		}
		
	} while ((bytes_parsed < con->buffer_len) && bytes_parsed && bytes_left);

	if (bytes_parsed == con->buffer_len)
		con_reset_buffer(con);

	/* gui update */
	gui_update_con_status(con);

 skip:	
	
	newtella_free(gh);
}

int newtella_sched(void)
{
	GList *iter;
	struct newtella_connection *con;

	gint now = time((time_t)NULL);

 	iter = g_list_first(gl_con);
	while (iter) {
		con = iter->data;

		if (!con)
			break;

		switch (con->state) {
		case GS_FAILED:
			/* remove failed connections */
			if (now - con->last_time > CON_REMOVE)
				con_remove(con);
			break;
		case GS_HANDSHAKE:
			if (now - con->last_time > 2*CON_REMOVE) 
				con_remove(con);
			break;
		case GS_TIMEOUT:
			/* remove timout connections */
			if (now - con->last_time > CON_REMOVE) 
				con_remove(con);
			break;
		case GS_WAITINGCONNECT:
			/* mark timeout connections */
			if (now - con->last_time > CON_REMOVE)  
				con_mark(con, GS_TIMEOUT);
			break;
		case GS_CONNECTED:	
			/* send ping to idle connections */
			if (now - con->last_time > CON_IDLE) {
				if (g_list_length(con->queue))
					con_mark(con, GS_TIMEOUT);
				else
					/* try to wake up */
					gnutella_send_ping(con, 1);
			}
			/* kill too bad connections */
			if (con->total_bad_received > (guint32)(con->total_handled/2))
				con_mark(con, GS_FAILED);
			break;
		default:
			break;
		}
	
		iter = g_list_next(iter);
	}

	if (gl_options->connected < gl_options->min_con && gl_options->auto_connect) 
		con_activate();
	
	return 1;
}