olfile.cpp

socket文件传输源码

			/** 1. send the hello message. */
			ret = s_reg_sem.send(s_con_arg, &s_op_head, out_buf, 0);
			if (-1 != ret) {
				s_con_arg->tot_send += ret; 
				s_reg_sem.recv = recv_helloack;
			}
			else{
				//close(s_con_arg->fd);
				fclose(s_con_arg->fp);
			}

			/** 2. receive helloack message. */
			ret = s_reg_sem.recv(s_con_arg, &s_op_head, in_buf, OP_HEAD_SIZE);
			s_con_arg->tot_recv += ret;
		}
		else
		{
			// server side:
			/** hello message is processed upper. */
			s_reg_sem.send = send_helloack;

			/** 2. send the hello ack message*/
			ret = s_reg_sem.send(s_con_arg, &s_op_head, out_buf, OP_HEAD_SIZE);
			s_con_arg->tot_send += ret;
		}

		if (-1 != ret && OP_SUCCESS == s_op_head.ret_val) {
			s_reg_sem.recv = recv_data;
			s_reg_sem.send = send_data;
			s_con_arg->cfm_mode = s_op_head.cfm_mode;
		}
		else{
			//close(s_con_arg->fd);
			fclose(s_con_arg->fp);
			return -1;
		}

		/** 3. reading data from remote peer */
		// in_buf format: operation head + payload.
		phead = in_buf;
		ppayload = in_buf + OP_HEAD_SIZE;
		s_con_arg->tot_send = 0;

		/** reading pieces of data from the net of remote peer
		   and put them to the gfin_buf (file buffer pool) then 
		   write the mass of data to the file.  
		*/
		char *pdyn_fin = fin_buf; // dynamic pointer to the current position in the file in buffer pool.
		int dyn_fin_len = 0; // current valid data size of the file in buffer pool.
		payload_len = 0;

		fprintf(stderr, "Starting transfer data...\n");

		/** receive data from remote peer and write them to 
			the sink file.
		*/
		while( -1 != ret ){
			in_len = MAX_MSG_LEN + OP_HEAD_SIZE; // desired receive length
			/** receive a data message from the remote peer, 
				and fill the s_op_head struct.
			*/
			if ( -1 == (in_len = s_reg_sem.recv(s_con_arg, &s_op_head, in_buf, in_len)))
			{
				fprintf(stderr, "\nReceive data message fail. Server may be exited abnormally\n");
				ret = -1;
				in_len = fwrite(fin_buf, sizeof(char), dyn_fin_len, s_con_arg->fp);
				show_transfer(s_con_arg);
				break;
			}
			
			payload_len = in_len - OP_HEAD_SIZE;
			s_con_arg->tot_recv += in_len;
			s_con_arg->tot_file_recv += payload_len;

			if (s_op_head.ret_val == OP_CMPOK) {
				ret = 0;
				//in_len = write(s_con_arg->fd, fin_buf, dyn_fin_len);
				in_len = fwrite(fin_buf, sizeof(char), dyn_fin_len, s_con_arg->fp);
				show_transfer(s_con_arg);
				break;
			}

			if (s_op_head.ret_val != OP_SUCCESS) {
				ret = -1;
				break;
			}

			/** send the dataack message to inform the remote peer a successful transfering of 
			   data message is completed, prepare for the next one. 
			*/
			if ( 0 <= payload_len ) {
				s_op_head.ret_val = OP_SUCCESS;
				if (-1 == send_dataack(s_con_arg, &s_op_head, out_buf, OP_HEAD_SIZE)) {
					ret = -1;
					break;
				};
				s_con_arg->tot_send += OP_HEAD_SIZE;
			}

			
			if (0 > payload_len ) {
				ret = -1;
				break;
			}

			if (0 == payload_len) {
				ret = 0;
				break;
			}

			/** put the piece of buffer into the in-bound file buffer pool 
				dyn_fin_len indicates current total data putted in.
			*/
			if (payload_len < (FILE_BUF_LEN - dyn_fin_len)) {
				// still has enough space for the payload.
				memcpy(pdyn_fin, ppayload, payload_len);
				pdyn_fin += payload_len;
				dyn_fin_len += payload_len;
			}
			else 
			{
				// write the received useful data to file opened.
				// actually this in_len stores the actual written bytes.
				//in_len = write(s_con_arg->fd, fin_buf, dyn_fin_len);
				in_len = fwrite(fin_buf, sizeof(char), dyn_fin_len, s_con_arg->fp);
				// put the remain data to the file buffer pool from the very beginning.
				pdyn_fin = fin_buf;
				dyn_fin_len = payload_len;
				memcpy(pdyn_fin, ppayload, dyn_fin_len);
				pdyn_fin += payload_len; // point to next start of segment.
				show_transfer(s_con_arg);
			}
		} // end of while 

//		if (0 == ret) {
			// operation completed successfully
			//close(s_con_arg->fd);
			fclose(s_con_arg->fp);
//		}
	}
	else if ( DATA_SRC == s_con_arg->con_mode ) {
		// the local peer is data source, perform a reading operation for the remote peer.
		if (CLIENT == s_con_arg->peer_mode) {
			s_reg_sem.send = send_hello;
			s_reg_sem.con = s_con_arg;
			s_op_head.op_type = OP_WRITE;
			s_op_head.ret_val = OP_SUCCESS;
			/** 1. send the hello message. */
			ret = s_reg_sem.send(s_con_arg, &s_op_head, out_buf, OP_HEAD_SIZE);
			if (-1 != ret) {
				s_con_arg->tot_send += ret;
				s_reg_sem.recv = recv_helloack;
			}
			else{
				//close(s_con_arg->fd);
				fclose(s_con_arg->fp);
			}
		
			/** 2. receive helloack message. */
			ret = s_reg_sem.recv(s_con_arg, &s_op_head, in_buf, OP_HEAD_SIZE);
			s_con_arg->tot_recv += ret;
		}
		else
		{
			// server side:
			/** the hello message is processed upper. */
			s_reg_sem.send = send_helloack;

			/** 2. send the hello ack message*/
			ret = s_reg_sem.send(s_con_arg, &s_op_head, out_buf, OP_HEAD_SIZE);
			s_con_arg->tot_send += ret;
		}

		if (-1 != ret && OP_SUCCESS == s_op_head.ret_val) {
			s_reg_sem.recv = recv_data;
			s_reg_sem.send = send_data;
		}
		else{
			//close(s_con_arg->fd);
			fclose(s_con_arg->fp);
			return -1;
		}

		/** 3. writing data to the remote peer */
		// in_buf format: operation head + payload.
		char *phead = out_buf;
		char *ppayload = out_buf + OP_HEAD_SIZE;
		s_op_head.msg_type = MSG_DATA;

		/** reading pieces of data from the file of local peer
		and put them to the gfout_buf (file buffer pool) then 
		write(send) the mass of data to the remote peer.  
		*/
		char *pdyn_fout = fout_buf; // dynamic pointer to the current position in the out-bound file buffer pool.
		int dyn_fout_len = 0; // current valid data size of the out-bound file buffer pool.
		int fout_len;	// one mass length read from the file.
		payload_len = 0;
		
		fprintf(stderr, "Starting transfer data...\n");
		while( -1 != ret ){
			// read a mass form file.
			//fout_len = read(s_con_arg->fd, fout_buf, FILE_BUF_LEN);
			fout_len = fread(fout_buf, sizeof(char), FILE_BUF_LEN, s_con_arg->fp);
			if ( fout_len < 0 ) {
				ret = -1;
				s_op_head.ret_val = OP_RDERR;
			}

			show_transfer(s_con_arg);

			if ( fout_len == 0) {
				ret = 0;
				s_op_head.ret_val = OP_CMPOK;
				// send the COMPK return value.
				s_con_arg->tot_send += OP_HEAD_SIZE;
				if ( -1 == (out_len = s_reg_sem.send(s_con_arg, &s_op_head, phead, OP_HEAD_SIZE)))
				{
					fprintf(stderr, "Send completion data message fail.\n");
					ret = -1;
				}
				break;
			}

			// send pieces of data
			pdyn_fout = fout_buf;
			dyn_fout_len = 0; // how many bytes has been sent out.
			while(fout_len > dyn_fout_len)
			{
				if (fout_len - dyn_fout_len > MAX_MSG_LEN ) 
					out_len = MAX_MSG_LEN;
				else
					out_len = fout_len - dyn_fout_len;

				memcpy(ppayload, pdyn_fout, out_len);
				/** 1. there could be a hidden trouble: originaly out_len buffer will be sent, 
				  but if uses send function and returns a value less than the out_len,
				  so the pointer should pull back to the gap of the out_len - value;
				*/
				// send data to remote peer
				if ( -1 == (out_len = s_reg_sem.send(s_con_arg, &s_op_head, phead, out_len + OP_HEAD_SIZE)))
				{
					fprintf(stderr, "Send data message fail.\n");
					ret = -1;
					break;
				}
				
				payload_len = out_len - OP_HEAD_SIZE;
				s_con_arg->tot_send += out_len;
				s_con_arg->tot_file_send += payload_len;
				// point to the next segment in the send file pool.
				/** 2. so the pointer added at an actural offset responds to the send data length. */
				pdyn_fout += payload_len;
				dyn_fout_len += payload_len;

				/** wait until the remote send back a dataack message to inform a successful data
				   message transferring.
				*/
				if (out_len >= 0) {
					if (-1 == recv_dataack(s_con_arg, &s_op_head, in_buf, OP_HEAD_SIZE)) {
						ret = -1;
						break;
					};
					s_con_arg->tot_recv += OP_HEAD_SIZE;
				}

				if (s_op_head.ret_val != OP_SUCCESS) {
					ret = -1;
					break;
				}
			} // end of while of send mass data
			
			if (s_op_head.ret_val != OP_SUCCESS) {
				ret = -1;
				break;
			}
		} 

//		if (0 == ret) {
			// operation completed successfully
			//close(s_con_arg->fd);
			fclose(s_con_arg->fp);
//		}
	}
	else {
		ret = -1;
	}

	return ret;
}

void show_err()
{
#ifdef WIN32
	int err = GetLastError();
	switch(err) {
	case WSAEHOSTUNREACH:
		fprintf(stderr, "Host unreachable.\n");
		break;
	case WSAECONNREFUSED:
		fprintf(stderr, "Connection refused on host. \nMaybe the service is inactive.\n");
		break;
	case WSAEADDRINUSE:
		fprintf(stderr, "Address already in use. ");
		break;
	default:
		fprintf(stderr, "Unspecfic error # %d.\n", err);
	}
#endif
}

void show_transfer(struct st_con_arg *s_con)
{
	time_t t_elaps, t_hou, t_min, t_sec;
	int samp_val;
	long fsize = s_con->fstat.st_size; 
	long lspeed;  // bytes/s
	float fspeed, fspeed2; // bytes/s
	float fcmp = 0.0;		// completed percent
	system(CLR_SCR);
	time(&t_elaps);
	t_elaps -= t_start;

	fprintf(stderr, "File %s, total %ld Bytes.\n", s_con->src_addr.ppath, fsize);
	fprintf(stderr, "-------------------------------------------------\n");
	fprintf(stderr, "Total send\tTotal recv\tTotal file");
	if (s_con->con_mode == DATA_SNK) {
		fprintf(stderr, " recv (Bytes)\n %8d\t%8d\t%8d\n", s_con->tot_send, s_con->tot_recv, s_con->tot_file_recv);	
		samp_val = s_con->tot_recv;
		fcmp = 100*((float)s_con->tot_file_recv/fsize);
	}
	else
	{
		fprintf(stderr, " send (Bytes)\n %8d\t%8d\t%8d\n", s_con->tot_send, s_con->tot_recv, s_con->tot_file_send);	
		samp_val = s_con->tot_send;
		fcmp = 100*((float)s_con->tot_file_send/fsize);
	}

	if (t_elaps!=0) {
		fprintf(stderr, "Data transfer speed: ");
		fspeed2 = (float)samp_val/t_elaps;
		if (s_con->tot_send<1024) {
			fspeed = fspeed2;
			fprintf(stderr, "%8.2f Bytes/s\n", fspeed);
		}
		else{
			fspeed = fspeed2/1024;
			fprintf(stderr, "%8.2f KB/s\n", fspeed);
		}
	}

	if ( fspeed2 > ftop_spd) {
		ftop_spd = fspeed2;
	}

	if (ftop_spd > 1024) 
		fprintf(stderr, "Top transfer speed: %8.2f KBytes/s\n", ftop_spd/1024);
	else
		fprintf(stderr, "Top transfer speed: %8.2f Bytes/s\n", ftop_spd);

	t_hou = t_elaps/3600;
	t_min = (t_elaps%3600)/60;
	t_sec = (t_elaps%3600)%60;
	fprintf(stderr, "Total time elapse %02dh:%02dm:%02ds\n", t_hou, t_min, t_sec);
	if (t_elaps>0) {
		lspeed = samp_val/t_elaps;
		if (lspeed>0) 
			t_elaps = (s_con->fstat.st_size - samp_val)/ lspeed;
		else
			t_elaps = 1;
	}

	t_hou = t_elaps/3600;
	t_min = (t_elaps%3600)/60;
	t_sec = (t_elaps%3600)%60;
	fprintf(stderr, "Total time remain %02dh:%02dm:%02ds\n", t_hou, t_min, t_sec);
	fprintf(stderr, "\t %3.2f%% Completed.\n", fcmp);
}

/** endif olfile.cpp */