head     1.1;
branch   ;
access   ;
symbols  ;
locks    ; strict;
comment  @@;


1.1
date     96.11.08.11.30.32;  author dejong;  state Exp;
branches ;
next     ;


desc
@Fixed bug in Tcp::recv at EOF return 0 (don't wait)
@



1.1
log
@Initial revision
@
text
@/*
   File: network.cc

   By: Alex Theo de Jong
   Created: April 1996

   Description:
   This file contains all network functions to be used with the mpeg2player. This
   is a temporary solution. The complete AtmSocket API should be used for this
   in order to use configuration/testing facilities. Hopefully we can do that im
   June/July 1996
*/

#ifdef __GNUG__
#pragma implementation
#endif

#include "athread.hh"

#include <stdio.h>
#include <stdlib.h>
#include <String.h>
#include <fstream.h>
#include <sys/time.h>
#include <sys/errno.h>
#include <fcntl.h>
#ifdef IRIX
#include <bstring.h>
#include <stropts.h>
#include <poll.h>
#endif

#include "error.hh"
#include "debug.hh"
#include "util.hh"

#include "network.hh"



// Socket File

SocketFile::SocketFile(const char* filen){
  TRACER("SocketFile::SocketFile(const char* filen)");
  filename=filen;
  /*
  file.open(filename.chars());
  if (!file){
    error("could not open file `" << filename << "' for reading");
  }
  */
  if ((socketfd=open(filename.chars(), O_RDONLY | O_NDELAY))<=0){
    error("could not open file `" << filename << "' for reading");
  }
}

SocketFile::~SocketFile(){
  TRACER("SocketFile::~SocketFile()");
  /*
  if (filename.length()){
    file.close();
  }
  */
  ::close(socketfd);
}


int SocketFile::recv(unsigned char* data, int size){
  DEBUGGER("int SocketFile::recv(unsigned char* data, int size)");
  bfr=data;
  read_bytes=0;
  do {
    /*
    if ((bytes=(file.read(data, size).gcount()))<=0)
    */
    if ((bytes=read(socketfd, data, size))<=0)
      return bytes;
    bfr+=bytes;
    read_bytes+=bytes;
  }
  while (read_bytes<size);
  return read_bytes;
};


int SocketFile::close(){
  TRACER("int SocketFile::close()");
  if (filename.length() && socketfd>0){
    return ::close(socketfd);
  }
  return -1;
}

// Socket TCP

SocketTcp::SocketTcp(unsigned int asap, unsigned int pdu_size){
  TRACER("SocketTcp::SocketTcp(unsigned int asap, unsigned int pdu_size)");

  sockaddr_in server_addr;
  
  if ((socketfd=::socket(AF_INET, SOCK_STREAM, IPPROTO_IP))<0){
    error("can't open stream socket");
    exit(1);
  }
  
#if (defined(IRIX) || defined(IRIX_PTHREAD))
  bzero(&server_addr, sizeof(server_addr));
#else
  memset((char*) &server_addr, 0, sizeof(server_addr));
#endif
  
#ifdef SOLARIS
  struct hostent* host=0;

  // Try ATM
  String name(getenv("HOSTNAME"));
  name+="-atm";  // default extension for ATM
  if (name.length()!=0 && (host=gethostbyname(name.chars()))==0){
    // if no ATM, try regular HOSTNAME address
    if ((host=gethostbyname(getenv("HOSTNAME")))==0){
      error("could not get host by name");
      athr_exit(0);
    }
  }

  server_addr.sin_addr.s_addr =((struct in_addr *) host->h_addr_list[0])->s_addr;
#endif
  
  server_addr.sin_family=AF_INET;
  server_addr.sin_port=htons(asap);
  while (1){
    if (::bind(socketfd, (struct sockaddr*) &server_addr, sizeof(server_addr))<0){
      ::close(socketfd);
      error("could not bind local address");
      exit(1);
    }
    else break;
    asap++;
  }
  
  int reuseaddr=1;
  if (setsockopt(socketfd, SOL_SOCKET, SO_REUSEADDR, (char*) &reuseaddr, sizeof(reuseaddr))<0){
    error("can't set sender socket option (SO_REUSEADDR)"); exit(1);
  }
  
  if (pdu_size){
    unsigned int read_size=pdu_size;
    if (setsockopt(socketfd,SOL_SOCKET,SO_SNDBUF,(char*)&read_size,sizeof(read_size))<0){
      error("can't set socket option (SO_SNDBUF)"); exit(1);
    }
    if (setsockopt(socketfd,SOL_SOCKET,SO_RCVBUF,(char*)&read_size,sizeof(read_size))<0){
      error("can't set socket option (SO_RCVBUF)"); exit(1);
    }
  }
  
  int alive=1;
  if (setsockopt(socketfd,SOL_SOCKET,SO_KEEPALIVE,(char*)&alive,sizeof(alive))<0){
    error("can't set socket option (SO_KEEPALIVE)"); exit(1);
  }
  int error=0, error_len=sizeof(error);
  if (getsockopt(socketfd, SOL_SOCKET, SO_ERROR, (char*) &error, &error_len)<0){
    error("can't get sender socket option (SO_ERROR)"); exit(1);
  }
  
  if (::listen(socketfd, 1)<0){
    error("could not listen to connection");
    exit(0);
  }
}

SocketTcp::~SocketTcp(){
  TRACER("SocketTcp::~SocketTcp");
  ::close(fd);
}

int SocketTcp::accept(){
  DEBUGGER("int SocketTcp::accept()");
  sockaddr_in client_addr;
  
  int client_len=sizeof(client_addr);
#if (defined(IRIX) || defined(IRIX_PTHREAD))
  bzero(&client_addr, sizeof(client_addr));
#else
  memset((char*) &client_addr, 0, sizeof(client_addr));
#endif
  
  if ((fd=::accept(socketfd, (struct sockaddr*) &client_addr, &client_len))<=0){
    error("could not accept connection (" << itoa(errno) << ")");
    athr_exit(0);
  }
  ::close(socketfd);
  message("received connection on TCP!");
  
  return fd;
}

int SocketTcp::recv(unsigned char* data, int size){
  DEBUGGER("int SocketTcp::recv(unsigned char* data, int size)");
  bfr=data;
  read_bytes=0;
  do {
    if ((bytes=::read(fd, bfr, size-read_bytes))<=0)
      return bytes;
    bfr+=bytes;
    read_bytes+=bytes;
  }
  while (read_bytes<size);
#ifdef DEEPDEBUG
  static int count=0;
  unsigned int test=((data[0]<<24) | (data[1] << 16) | (data[2]<<8) | data[3]);  printf("First 4 bytes=%x\n", test);
  test=((data[188]<<24) | (data[189] << 16) | (data[190]<<8) | data[191]); 
  printf("First 4 bytes=%x\n", test);
  if (count++==10) exit(0);
#endif
  return read_bytes;
};


int SocketTcp::close(){
  int err=::close(fd);
  TRACER("int SocketTcp::close()");
  return err;
}


// Socket UDP

SocketUdp::SocketUdp(unsigned int asap, unsigned int pdu_size){
  TRACER("SocketUdp::SocketUdp()");

  sockaddr_in server_addr;
  
  if ((socketfd=::socket(AF_INET, SOCK_DGRAM, IPPROTO_IP))<0){
    error("can't open dgram socket");
    exit(1);
  }
  
#if (defined(IRIX) || defined(IRIX_PTHREAD))
  bzero(&server_addr, sizeof(server_addr));
#else
  memset((char*) &server_addr, 0, sizeof(server_addr));
#endif
  
#ifdef SOLARIS
  struct hostent* host=0;
  //  if (option==1){
  if ((host=gethostbyname(getenv("HOSTNAME")))==0){
    error("could not get host by name");
    athr_exit(0);
  }

  /*
  else {
    String name(getenv("HOSTNAME"));
    name+="-atm";  // default extension for ATM
    if (name.length()!=0 && (host=gethostbyname(name.chars()))==0){
      error("could not get host by name");
      athr_exit(0);
    }
  }
  */

  server_addr.sin_addr.s_addr =((struct in_addr *) host->h_addr_list[0])->s_addr;
#endif
 
  server_addr.sin_family=AF_INET;
  server_addr.sin_port=htons(asap);
  while (1){
    if (::bind(socketfd, (struct sockaddr*) &server_addr, sizeof(server_addr))<0){
      ::close(socketfd);
      error("could not bind local address");
      exit(1);
    }
    else break;
    asap++;
  }
  
  int reuseaddr=1;
  if (setsockopt(socketfd, SOL_SOCKET, SO_REUSEADDR, (char*) &reuseaddr, sizeof(reuseaddr))<0){
    error("can't set sender socket option (SO_REUSEADDR)"); exit(1);
  }
 
  if (pdu_size){
    unsigned int read_size=pdu_size;
    if (setsockopt(socketfd,SOL_SOCKET,SO_SNDBUF,(char*)&read_size,sizeof(read_size))<0){
      error("can't set socket option (SO_SNDBUF)"); exit(1);
    }
    if (setsockopt(socketfd,SOL_SOCKET,SO_RCVBUF,(char*)&read_size,sizeof(read_size))<0){
      error("can't set socket option (SO_RCVBUF)"); exit(1);
    }
  }
  
  int alive=1;
  if (setsockopt(socketfd,SOL_SOCKET,SO_KEEPALIVE,(char*)&alive,sizeof(alive))<0){
    error("can't set socket option (SO_KEEPALIVE)"); exit(1);
  }
  int error=0, error_len=sizeof(error);
  if (getsockopt(socketfd, SOL_SOCKET, SO_ERROR, (char*) &error, &error_len)<0){
    error("can't get sender socket option (SO_ERROR)"); exit(1);
  }
 
  pcli_addr=(sockaddr*) &cli_addr;
  cli_len=sizeof(cli_addr);
}

SocketUdp::~SocketUdp(){
  TRACER("SocketUdp::~SocketUdp");
  ::close(fd);
}

int SocketUdp::accept(){
  TRACER("int SocketUdp::accept()");
  fd=socketfd;
  message("received connection on UDP!");
  return fd;
}

int SocketUdp::recv(unsigned char* data, int size){
  DEBUGGER("int SocketUdp::recv(unsigned char* data, int size)");
  bfr=data;
  read_bytes=0;
  do {
    if ((bytes=::recvfrom(fd, (char*) bfr, size-read_bytes, 0, pcli_addr, &cli_len))==size)
      return bytes;
    bfr+=bytes;
    read_bytes+=bytes;
  } 
  while (read_bytes<size);

  return read_bytes;
}

int SocketUdp::close(){
  int err=::close(fd);
  TRACER("int SocketUdp::close()");
  return err;
}



#ifdef FORE_ATM

// Socket ATM SPANS

SocketAtmSpans::SocketAtmSpans(unsigned int sap) : asap(sap) {
  TRACER("SocketAtmSpans::SocketAtmSpans(unsigned int sap)");

  Atm_info info;
  aal_type=aal_type_5;
  dataflow=duplex;
  
  if ((socketfd=atm_open(ATMDEVICE, O_RDWR, &info))<0){
    atm_error("atm_open : ");
    exit(1);
  }
  
  int queue_size=1;
  if (atm_bind(socketfd, asap, &asap, queue_size)<0){
    atm_error("atm_bind");
    exit(1);
  }
}

SocketAtmSpans::~SocketAtmSpans(){
  TRACER("SocketAtmSpans::~SocketAtmSpans()");
  ::atm_close(fd);
}

int SocketAtmSpans::accept(){
  TRACER("int SocketAtmSpans::accept()");
  Atm_qos qos_server;
  int conn_id(0);
  
  if (atm_listen(socketfd, &conn_id, &atm_endpoint, &qos_server, &aal_type) < 0){
    atm_error("atm_listen");
    exit(1);
  }
  qos.peak_bandwidth.target  = qos_server.peak_bandwidth.target; 
  qos.peak_bandwidth.minimum = qos_server.peak_bandwidth.minimum;
  qos.mean_bandwidth.target  = qos_server.mean_bandwidth.target;
  qos.mean_bandwidth.minimum = qos_server.mean_bandwidth.minimum;
  qos.mean_burst.target      = qos_server.mean_burst.target;
  qos.mean_burst.minimum     = qos_server.mean_burst.minimum;
  
  if (atm_accept(socketfd, socketfd, conn_id, &qos, dataflow) < 0) {
    atm_error("atm_accept");
    exit(1);
  }
  fd=socketfd;
  message("received connection on ATM SPANS!");

  return fd;
}

int SocketAtmSpans::recv(unsigned char* data, int size){
  DEBUGGER("int SocketAtmSpans::recv(unsigned char* data, int size)");
  bfr=data;
  read_bytes=0;
  do {
    if ((bytes=::atm_recv(fd, (caddr_t) bfr, size-read_bytes))<=0)
      // ATM returns -1 when connection is terminated
      return (bytes==-1 || bytes==0) ? 0 : bytes;
    bfr+=bytes;
    read_bytes+=bytes;
  }
  while (read_bytes<size);

  return read_bytes;
}

int SocketAtmSpans::close(){
  int err=::atm_close(fd);
  TRACER("int SocketAtmSpans::close()");
  return err;
}



// class ATM PVC

SocketAtmPvc::SocketAtmPvc(unsigned int vci, unsigned int vpi){
  TRACER("SocketAtmPvc::SocketAtmPvc(unsigned int vci, unsigned int vpi)");

  Atm_info info;
  Atm_conn_resource  qos;
  aal_type=aal_type_5;

  if ((socketfd=atm_open(ATMDEVICE, O_RDWR, &info))<0){
    atm_error("atm_open: ");
    exit(1);
  }

  /*
  if (info.mtu < pdu_size){
    error("sdu size to big for maximum ATM mtu");
    exit(1);
  }
  */

  qos.peak_bandwidth =  50000;  // set some dummy values
  qos.mean_bandwidth =  500000;

  vpvc= (vpi  & 0x00000FFF) << 16;
  vpvc+=(vci & 0x0000FFFF);

  if (atm_bind_pvc(socketfd, vpvc, aal_type, &qos)!=0){
    atm_error("atm_bind_pvc");
    exit(1);
  }
}


SocketAtmPvc::~SocketAtmPvc(){
  TRACER("SocketAtmPvc::~SocketAtmPvc()");
  ::atm_close(fd);
}

int SocketAtmPvc::accept(){
  TRACER("int SocketAtmPvc::accept()");

  fd=socketfd;  // ready to receive data
  message("received connection on ATM PVC!");

  return fd;
}

int SocketAtmPvc::recv(unsigned char* data, int size){
  DEBUGGER("int SocketAtmPvc::recv(unsigned char* data, int size)");
  bfr=data;
  read_bytes=0;
  do {
    if ((bytes=::atm_recvfrom(fd, (caddr_t) bfr, size-read_bytes, &vpvc))<=1){
      if (bytes==1) return 0; // eof   EOF protocol if using a PVC
      return bytes;
    }
    bfr+=bytes;
    read_bytes+=bytes;
  }
  while (read_bytes<size);
  return read_bytes;
}
 
int SocketAtmPvc::close(){
  int err=::atm_close(fd);
  TRACER("int SocketAtmPvc::close()");
  return err;
}

#endif // FORE_ATM


// SocketMulti waits at several ports and accepts the one first activated

SocketMulti::SocketMulti(unsigned int asap, int pdu_size, unsigned int vci, unsigned int vpi){
  TRACER("SocketMulti::SocketMulti(unsigned int asap, int pdu_size, vci, int vpi)");

  sockets=new Socket*[SOCKETMULTI_MAX];
  count=0;
  sockets[count++]=new SocketTcp(asap, pdu_size);
  sockets[count++]=new SocketUdp(asap, pdu_size);
#ifdef FORE_ATM
  // test to find ATM board
  Atm_info info;
  int dummy;
  if ((dummy=atm_open(ATMDEVICE, O_RDWR, &info))>0){
    atm_close(dummy);
    // found an ATM board; thus include these while wating for connection
    sockets[count++]=new SocketAtmSpans(asap);
    sockets[count++]=new SocketAtmPvc(vci, vpi);
  }
  sockets[count]=0;
#endif 
}


SocketMulti::SocketMulti(const char* name){ 
  TRACER("SocketMulti::SocketMulti(const char* name)");
  
  sockets=0;
  socket=new SocketFile(name);
  count=1;
}

SocketMulti::~SocketMulti(){
  if (sockets){
    for (int i=0; sockets[i]!=0; i++) 
      delete sockets[i];
    delete[] sockets;
  }
  else delete socket;
}

int SocketMulti::accept(){
  TRACER("int SocketMulti::accept()");
  if (sockets==0) return 0;

  pollfd fds[SOCKETMULTI_MAX];
  for (int i=0; i<count; i++){
    fds[i].fd=sockets[i]->socketfd;
    fds[i].events= POLLRDNORM | POLLPRI; //  POLLIN | POLLRDBAND | );
  }

  message("waiting for connection ... ");

  if (poll(fds, count, -1)<0){
    error("could not call poll");
    exit(1);
  }

  for (int j=0; j<count; j++){
    if (fds[j].revents>0){
      return (socket=sockets[j])->accept();
    }
  }

  error("failed to find event after poll");
  return -1;
}


@