ci.c

一个很好的例子-基于Linux的消息传输系统

#include"ci.h"
void * SendDataMsgCli(SendPack * mySendPack)
{
   Buffer * mybuffer = createBuffer();
   int fd,numbytes,i=0;
  
   struct sockaddr_in server;

   strToByte(mySendPack->msg,mybuffer);

   //inet_aton(mySendPack->ip,&server.sin_addr);

   if((fd=socket(AF_INET,SOCK_STREAM,0))==-1)
     {
       printf("socket error");
       exit(1);
     }
   bzero(&server,sizeof(server));
   inet_aton(mySendPack->ip,&server.sin_addr);
   server.sin_family=AF_INET;
   server.sin_port=htons(mySendPack->port);

   if(connect(fd,(struct sockaddr*)&server,sizeof(struct sockaddr))==-1)
      {
        printf("connect error\n");
        exit(1);
      }
    send(fd,mybuffer->bytes,mybuffer->length,0);
    destroyBuffer(mybuffer);
    close(fd);
}

void * SendDataMsgSer(void * ref1)
{    
   CI * ci = ( CI *)ref1;
   char * ip = "111";      //this ip no meaning
  
   int fd,numbytes,port;
   char buf[MAXDATASIZE];
   int portnum=ci->listenPort;
   int listenfd,connectfd;
   struct sockaddr_in server;
   struct sockaddr_in client;
   int sin_size;
    Buffer * mybuffer;
    SendPack *sendBuf;
    Message * mymessage;
    port =8888;     // this port no meaning
   if((listenfd=socket(AF_INET,SOCK_STREAM,0))==-1)
      {
         perror("created socket failed");
         exit(1);
      }
   int opt=SO_REUSEADDR;
   setsockopt(listenfd,SOL_SOCKET,SO_REUSEADDR,&opt,sizeof(opt));

   bzero(&server,sizeof(server));
   server.sin_family=AF_INET;
   server.sin_port=htons(portnum);
   server.sin_addr.s_addr=htonl(INADDR_ANY);

   if(bind(listenfd,(struct sockaddr *)&server,sizeof(struct sockaddr))==-1)
     {
       perror("bind error");
       exit(1);
     }

    if(listen(listenfd,BACKLOG)==-1)
       {
         perror("listen error");
         exit(1);
       }
    printf("i am listening the cleint...........\n");
    while(1)
    {
     mybuffer = createBuffer();

     sin_size=sizeof(struct sockaddr);
     if((connectfd=accept(listenfd,(struct sockaddr*)&client,&sin_size))==-1)
        {
                perror("accept error");
                exit(1);
                }
      printf("i get a connection from:%s\n",inet_ntoa(client.sin_addr));

     if((numbytes=recv(connectfd,buf,MAXDATASIZE,0))==-1)
      {
         printf("recv() error\n");
         exit(1);
      }
   buf[numbytes]='\0';
   mybuffer->bytes = buf;
   mymessage = byteToStruct (mybuffer);
 
   while(1)
   {
   if(ci->receiveBuffer->head!= NULL && ci->receiveBuffer->length == 1000) 
        {
          sleep(0.5);   
        }                         
   else
     break;
   }
   putMessageRec(ci,mymessage,ip,port);

   close(connectfd);
 //  free(mymessage);
   
  // destroyBuffer(mybuffer);  
  // destroyMessage(mymessage);
  }
      close(listenfd);
}
void * midSend(void * bb)
{ 

    CI * myci = ( CI *)bb;
   while(1)
   {  
       SendPack * mySendPack2;     
       
       while(myci->sendBuffer->head== NULL)
      {
              sleep(1);
      }
       mySendPack2 =(SendPack* )pop(myci->sendBuffer,myci->sendBuffer->head);
//       printf("pop msg type : %d\n",mySendPack2->msg->type);  
       SendDataMsgCli(mySendPack2);
   }  
}
void * midRec(void * cc)
{
   SendDataMsgSer(cc);
}
    
 
void StartCI( CI * ci2)
{
         pthread_create(&td2,NULL,midRec,ci2);
         pthread_create(&td1,NULL,midSend,ci2 );      
}
 
Message * getMsgFromCi(CI *ci)
{   
    SendPack * mySendPack ;
    Message * message;
    while(ci->receiveBuffer->head==NULL)
    {  
       sleep(1);
    }
    mySendPack = (SendPack *)pop(ci->receiveBuffer,ci->receiveBuffer->head);
    message = mySendPack->msg;
    return message;
}