ms_tcp.c

1. 8623L平台

/*
 *
 * Copyright (c) Sigma Designs, Inc. 2005. All rights reserved.
 *
 */
#define ALLOW_OS_CODE 1
#include "rmdef/rmdef.h"
#include "rmcore/include/rmascii.h"
#include <stdio.h>
#include "ms_tcp.h"
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <string.h>
#include <errno.h>
#include <netdb.h>
#include <time.h>
#include <sys/poll.h>


/* Maximum number of bytes we can read at once from the server */
#define MAX_BUFFER_SIZE (8*1024)
static RMuint8 tcp_buffer[MAX_BUFFER_SIZE];

struct ms_tcp_socket_s {
	int sockfd;
	struct sockaddr_in name;
};


/**
 * Create a new TCP socket, ready for sending data to url
 *
 * @param url- http url http://host:port/xxx
 * @return a valid ms_tcp_socket_t, or NULL on error
 */
ms_tcp_socket_t * open_ms_tcp_socket(RMascii *url)
{
	ms_tcp_socket_t *ms_socket = NULL;
	RMascii *purl = NULL;
	RMuint16 port;
	RMascii *host;
	RMascii *port_string;
	RMascii *filename;
	struct hostent *hostinfo;
	int status;

	ms_socket = (ms_tcp_socket_t *)RMMalloc(sizeof(ms_tcp_socket_t));
	if (ms_socket == NULL)
		return NULL;

	ms_socket->sockfd = socket(PF_INET, SOCK_STREAM, 0);
	if (ms_socket->sockfd < 0){
		RMDBGLOG((ENABLE,"Error creating socket : %s\n", strerror(errno)));
		goto error;
	}

	purl = RMMallocAndDuplicateAscii(url);
	if (purl == NULL)
		goto error;

	if (RMFindAsciiString(purl, "://", &host))
		host+=3;
	else
		goto error;

	if (RMFindAsciiCharacter(host, ':', &port_string)){
		port_string[0] = 0;
		port_string++;
		if (RMFindAsciiCharacter(port_string,'/', &filename))
			filename[0] = 0;
		else 
			goto error;
		RMasciiToUInt16(port_string, &port);
		port = htons(port);
	} else
		goto error;

	ms_socket->name.sin_family = AF_INET;
	ms_socket->name.sin_port = port;
	hostinfo = gethostbyname (host);
	if (hostinfo == NULL)
	{
		RMDBGLOG((ENABLE, "Error getting host url for %s : %s\n", host, strerror(errno)));
		goto error;
	}
	ms_socket->name.sin_addr = *(struct in_addr *) hostinfo->h_addr;

	status = connect(ms_socket->sockfd, (struct sockaddr *)&(ms_socket->name), sizeof(ms_socket->name));
	if (status < 0){
		RMDBGLOG((ENABLE, "Error connecting the socket : %s\n", strerror(errno)));
		goto error;
	}

	RMFree(purl);
	return ms_socket;

error :
	if (ms_socket)
		RMFree(ms_socket);

	if (purl)
		RMFree(purl);

	return NULL;
}
	
	
/**
 * Close a TCP connection, and free associated resources
 * 
 * @param ms_socket - the socket to close
 */
void close_ms_tcp_socket(ms_tcp_socket_t *ms_socket)
{
	int status;
	
	if (ms_socket == NULL)
		return;

	status = shutdown(ms_socket->sockfd, SHUT_RDWR);
	if (status < 0)
		RMDBGLOG((ENABLE,"Error closing MS socket : %s\n", strerror(errno)));

	RMFree(ms_socket);
}
	
/**
 * Send data on a TCP socket and wait for an anser for a given time.
 * This function is not reentrant, the answer should not be freed and
 * a subsequent call to ms_tcp_send_receive will modify it.
 *
 * @param ms_socket - ms_socket to use
 * @param message
 * @param message_size 
 * @param answer - answer received
 * @param answer_size - size of the answer
 * @param delay_s - global delay (s)
 * @return RM_OK on success, RM_ERROR if no message received in the given delay
 */
RMstatus ms_tcp_send_receive(
		ms_tcp_socket_t *ms_socket, 
		RMuint8 *message, RMuint32 message_size,
		RMuint8 **answer, RMuint32 *answer_size, 
		RMuint32 delay_s)
{
	ssize_t sent_bytes;
	time_t end_time;
	struct pollfd mpoll;
	int status;
	RMuint32 offset_tcp_buffer = 0;
	time_t small_dealy_ms;

	if (ms_socket == NULL ||
	    message == NULL ||
	    message_size == 0)
		return RM_ERROR;


	/* Try sending the message */
	RMDBGLOG((ENABLE,"Send %ld bytes message\n", message_size));
	sent_bytes = send(ms_socket->sockfd, (void *)message, (size_t)message_size, 0);
	if (sent_bytes < 0 ){
		RMDBGLOG((ENABLE,"Error sending data : %s\n", strerror(errno)));
		goto error;
	} else if ((unsigned) sent_bytes != message_size) {
		RMDBGLOG((ENABLE,"Error, only %ld/%ld bytes sent\n", sent_bytes, message_size));
		goto error;
	}

	
	end_time = time(NULL) + (time_t) delay_s;
	while ((end_time > time(NULL)) && (offset_tcp_buffer < MAX_BUFFER_SIZE)){
		/* Wait for more data */
		mpoll.fd = ms_socket->sockfd;
		mpoll.events = POLLIN | POLLPRI;
		small_dealy_ms = (end_time - time(NULL)) * 1000;
		if (small_dealy_ms < 0)
			small_dealy_ms = 0;
		status = poll(&mpoll, 1, (int) small_dealy_ms);
		if (status < 0 ){
			RMDBGLOG((ENABLE,"Error polling socket : %s\n", strerror(errno)));
			goto error;
		} else if ( status == 0 ) {
			RMDBGLOG((ENABLE,"Poll timed out ...\n"));
			continue;
		}

		/* Something happened in time */
		if (mpoll.revents & (POLLIN|POLLPRI)){
			int nread;
			
			nread = recv(ms_socket->sockfd, tcp_buffer+offset_tcp_buffer, MAX_BUFFER_SIZE - offset_tcp_buffer, MSG_DONTWAIT);
			if (nread < 0){
				RMDBGLOG((ENABLE,"Error reading MS socket : %s\n", strerror(errno)));
				goto error;
			} else if (nread == 0) {
				/* End of stream */
				break;
			}
			
			offset_tcp_buffer += nread;
			RMDBGLOG((ENABLE,"Received %ld bytes from MS server\n", nread));
		} else {
			RMDBGLOG((ENABLE,"Errorm, unexpected event happened\n"));
			goto error;
		}
	}

	*answer = tcp_buffer;
	*answer_size = offset_tcp_buffer;
	return RM_OK;
error:
	return RM_ERROR;
}