00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022
00023 #ifndef _GRTCPSINK_H_
00024 #define _GRTCPSINK_H_
00025
00026 #include <vector>
00027 #include <sys/socket.h>
00028 #include <netinet/in.h>
00029 #include <arpa/inet.h>
00030 #include <netdb.h>
00031 #include <fcntl.h>
00032 #include <stdio.h>
00033 #include <errno.h>
00034
00035 using std::vector;
00036
00037
00038
00039
00040
00041 static inline unsigned long int _local_htonl (unsigned long int hostlong)
00042 {
00043 return htonl (hostlong);
00044 }
00045
00046 static inline unsigned short int _local_htons (unsigned short int hostshort)
00047 {
00048 return htons (hostshort);
00049 }
00050
00051 template <class iType>
00052 class GrTCPSink : public VrSink<iType>
00053 {
00054 public:
00055 GrTCPSink (int port);
00056 virtual ~GrTCPSink ();
00057
00058
00059 virtual const char *name () { return "GrTCPSink"; }
00060 virtual int work3(VrSampleRange output,
00061 VrSampleRange inputs[], void *i[]);
00062
00063 protected:
00064 int listen_socket;
00065 std::vector<int> connection;
00066
00067 };
00068
00069 template <class iType>
00070 GrTCPSink<iType>::GrTCPSink (int port)
00071 {
00072 struct sockaddr_in server_addr;
00073
00074
00075 listen_socket = socket (AF_INET, SOCK_STREAM, 0);
00076 if (listen_socket == -1){
00077 perror ("GrTPCSink: socket");
00078 exit (1);
00079 }
00080
00081 memset (&server_addr, 0, sizeof (server_addr));
00082 server_addr.sin_family = AF_INET;
00083 server_addr.sin_addr.s_addr = _local_htonl (INADDR_ANY);
00084 server_addr.sin_port = _local_htons (port);
00085
00086 if (bind (listen_socket, (struct sockaddr *) &server_addr,
00087 sizeof (server_addr)) == -1){
00088 perror ("GrTCPSink: bind");
00089 exit (1);
00090 }
00091
00092 if (listen (listen_socket, 5) == -1){
00093 perror ("GrTCPSink: listen");
00094 exit (1);
00095 }
00096
00097 long file_flags = fcntl (listen_socket, F_GETFL);
00098 if (fcntl (listen_socket, F_SETFL, file_flags | O_NONBLOCK) == -1){
00099 perror ("GrTCPSink: fcntl / O_NONBLOCK");
00100 exit (1);
00101 }
00102 }
00103
00104 template<class iType> int
00105 GrTCPSink<iType>::work3(VrSampleRange output,
00106 VrSampleRange inputs[], void *ai[])
00107 {
00108 iType *in = ((iType **)ai)[0];
00109
00110
00111
00112
00113
00114
00115
00116
00117
00118 do {
00119
00120
00121
00122
00123
00124 int new_socket = accept (listen_socket, 0, 0);
00125 if (new_socket == -1){
00126 if (errno == EAGAIN){
00127
00128 }
00129 else {
00130 perror ("GrTCPSink: accept");
00131 exit (1);
00132 }
00133 }
00134 else {
00135 fprintf (stderr, "GrTCPSink: new connection\n");
00136 connection.push_back (new_socket);
00137 }
00138
00139
00140
00141
00142
00143 } while (connection.size () == 0);
00144
00145
00146
00147
00148
00149
00150 if (connection.size () == 0)
00151
00152 return output.size;
00153
00154
00155 long nbytes_to_write = output.size * sizeof (iType);
00156
00157
00158
00159 vector<int>::iterator p = connection.begin ();
00160
00161 while (p != connection.end ()){
00162
00163
00164
00165 long r;
00166 int flags = 0;
00167 #ifdef MSG_NOSIGNAL
00168 flags = MSG_NOSIGNAL;
00169 #endif
00170 if ((r = send (*p, in, nbytes_to_write, flags)) != nbytes_to_write){
00171 if (r == -1 && errno == EPIPE){
00172
00173 fprintf (stderr, "GrTCPSink: closing connection\n");
00174 close (*p);
00175 p = connection.erase (p);
00176 }
00177 else {
00178 perror ("GrTCPSink: send");
00179 }
00180 }
00181 else
00182 p++;
00183 }
00184
00185 return output.size;
00186 }
00187
00188
00189 template <class iType>
00190 GrTCPSink<iType>::~GrTCPSink ()
00191 {
00192 vector<int>::iterator p = connection.begin ();
00193 while (p != connection.end ()){
00194 close (*p);
00195 p++;
00196 }
00197
00198 close (listen_socket);
00199 }
00200
00201
00202 #endif // _GRTCPSINK_H_