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_