/* tcpreceive.c
* copyright © 2010-2015 IOhannes m zmölnig, IEM
* copyright (c) 2006-2010 Martin Peach
* copyright (c) Miller Puckette
*/
/* */
/* A client for unidirectional communication from within Pd. */
/* */
/* This program is free software; you can redistribute it and/or */
/* modify it under the terms of the GNU General Public License */
/* as published by the Free Software Foundation; either version 2 */
/* of the License, or (at your option) any later version. */
/* */
/* This program is distributed in the hope that it will be useful, */
/* but WITHOUT ANY WARRANTY; without even the implied warranty of */
/* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the */
/* GNU General Public License for more details. */
/* */
/* You should have received a copy of the GNU General Public License */
/* along with this program; if not, see */
/* http://www.gnu.org/licenses/ */
/* */
#define DEBUGLEVEL 1
static const char*objName = "tcpreceive";
#include "iemnet.h"
#ifndef _WIN32
/* needed for TCP_NODELAY */
# include <netinet/tcp.h>
#endif
#include <string.h>
/* ----------------------------- tcpreceive ------------------------- */
static t_class *tcpreceive_class;
#define MAX_CONNECTIONS 128 /* this is going to cause trouble down the line...:( */
typedef struct _tcpconnection {
long addr;
unsigned short port;
int socket;
struct _tcpreceive*owner;
t_iemnet_receiver*receiver;
} t_tcpconnection;
typedef struct _tcpreceive {
t_object x_obj;
t_outlet*x_msgout;
t_outlet*x_addrout;
t_outlet*x_connectout;
t_outlet*x_statusout;
int x_connectsocket;
int x_port;
int x_serialize;
int x_nconnections;
t_tcpconnection x_connection[MAX_CONNECTIONS];
t_iemnet_floatlist*x_floatlist;
} t_tcpreceive;
/* forward declarations */
static int tcpreceive_disconnect(t_tcpreceive *x, int id);
static int tcpreceive_find_socket(t_tcpreceive *x, int fd)
{
int i;
for (i = 0; i < MAX_CONNECTIONS; ++i)
if (x->x_connection[i].socket == fd) {
return i;
}
return -1;
}
static void tcpreceive_read_callback(void *w, t_iemnet_chunk*c)
{
t_tcpconnection*y = (t_tcpconnection*)w;
t_tcpreceive*x = NULL;
int index = -1;
if(NULL == y || NULL == (x = y->owner)) {
return;
}
index = tcpreceive_find_socket(x, y->socket);
if(index >= 0) {
if(c) {
/* TODO?: outlet info about connection */
/* gets destroyed in the dtor */
x->x_floatlist = iemnet__chunk2list(c, x->x_floatlist);
iemnet__streamout(x->x_msgout, x->x_floatlist->argc, x->x_floatlist->argv,
x->x_serialize);
} else {
/* disconnected */
tcpreceive_disconnect(x, index);
}
}
}
/* tcpreceive_addconnection tries to add the socket fd to the list */
/* returns 1 on success, else 0 */
static int tcpreceive_addconnection(t_tcpreceive *x, int fd, long addr,
unsigned short port)
{
int i;
for (i = 0; i < MAX_CONNECTIONS; ++i) {
if (x->x_connection[i].socket == -1) {
x->x_connection[i].socket = fd;
x->x_connection[i].addr = addr;
x->x_connection[i].port = port;
x->x_connection[i].owner = x;
x->x_connection[i].receiver =
iemnet__receiver_create(fd,
x->x_connection+i,
tcpreceive_read_callback,
0);
return 1;
}
}
return 0;
}
/* tcpreceive_connectpoll checks for incoming connection requests on the original socket */
/* a new socket is assigned */
static void tcpreceive_connectpoll(t_tcpreceive *x, int fd)
{
struct sockaddr_in from;
socklen_t fromlen = sizeof(from);
long addr;
unsigned short port;
if(fd != x->x_connectsocket) {
iemnet_log(x, IEMNET_FATAL, "callback received for socket:%d on listener for socket:%d", fd, x->x_connectsocket);
return;
}
fd = accept(fd, (struct sockaddr *)&from, &fromlen);
if (fd < 0) {
iemnet_log(x, IEMNET_ERROR, "could not accept new connection");
sys_sockerror("accept");
} else {
/* get the sender's ip */
addr = ntohl(from.sin_addr.s_addr);
port = ntohs(from.sin_port);
if (tcpreceive_addconnection(x, fd, addr, port)) {
x->x_nconnections++;
iemnet__numconnout(x->x_statusout, x->x_connectout, x->x_nconnections);
iemnet__addrout(x->x_statusout, x->x_addrout, addr, port);
} else {
iemnet_log(x, IEMNET_ERROR, "too many connections");
iemnet__closesocket(fd, 1);
}
}
}
static int tcpreceive_disconnect(t_tcpreceive *x, int id)
{
if(id >= 0 && id < MAX_CONNECTIONS && x->x_connection[id].port>0) {
iemnet__receiver_destroy(x->x_connection[id].receiver, 0);
x->x_connection[id].receiver = NULL;
iemnet__closesocket(x->x_connection[id].socket, 1);
x->x_connection[id].socket = -1;
x->x_connection[id].addr = 0L;
x->x_connection[id].port = 0;
x->x_nconnections--;
iemnet__numconnout(x->x_statusout, x->x_connectout, x->x_nconnections);
return 1;
}
return 0;
}
/* tcpreceive_closeall closes all open sockets and deletes them from the list */
static void tcpreceive_disconnect_all(t_tcpreceive *x)
{
int i;
for (i = 0; i < MAX_CONNECTIONS; i++) {
tcpreceive_disconnect(x, i);
}
}
/* tcpreceive_removeconnection tries to delete the socket fd from the list */
/* returns 1 on success, else 0 */
static int tcpreceive_disconnect_socket(t_tcpreceive *x, int fd)
{
int i;
for (i = 0; i < MAX_CONNECTIONS; ++i) {
if (x->x_connection[i].socket == fd) {
return tcpreceive_disconnect(x, i);
}
}
return 0;
}
static void tcpreceive_port(t_tcpreceive*x, t_floatarg fportno)
{
static t_atom ap[1];
int portno = fportno;
struct sockaddr_in server;
socklen_t serversize = sizeof(server);
int sockfd = x->x_connectsocket;
int intarg;
memset(&server, 0, sizeof(server));
SETFLOAT(ap, -1);
if(x->x_port == portno) {
return;
}
/* cleanup any open ports */
if(sockfd >= 0) {
sys_rmpollfn(sockfd);
iemnet__closesocket(sockfd, 1);
x->x_connectsocket = -1;
x->x_port = -1;
}
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if(sockfd<0) {
iemnet_log(x, IEMNET_ERROR, "unable to create socket");
sys_sockerror("socket");
return;
}
/* ask OS to allow another Pd to reopen this port after we close it. */
#ifdef SO_REUSEADDR
intarg = 1;
if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR,
(char *)&intarg, sizeof(intarg))
< 0) {
iemnet_log(x, IEMNET_ERROR, "unable to enable address re-using");
sys_sockerror("setsockopt:SO_REUSEADDR");
}
#endif /* SO_REUSEADDR */
#ifdef SO_REUSEPORT
intarg = 1;
if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEPORT,
(char *)&intarg, sizeof(intarg))
< 0) {
iemnet_log(x, IEMNET_ERROR, "unable to enable port re-using");
sys_sockerror("setsockopt:SO_REUSEPORT");
}
#endif /* SO_REUSEPORT */
/* Stream (TCP) sockets are set NODELAY */
intarg = 1;
if (setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY,
(char *)&intarg, sizeof(intarg)) < 0) {
iemnet_log(x, IEMNET_ERROR, "unable to enable immediate sending");
sys_sockerror("setsockopt:TCP_NODELAY");
}
server.sin_family = AF_INET;
server.sin_addr.s_addr = INADDR_ANY;
server.sin_port = htons((u_short)portno);
/* name the socket */
if (bind(sockfd, (struct sockaddr *)&server, serversize) < 0) {
iemnet_log(x, IEMNET_ERROR, "couldn't bind socket");
sys_sockerror("bind");
iemnet__closesocket(sockfd, 1);
sockfd = -1;
outlet_anything(x->x_statusout, gensym("port"), 1, ap);
return;
}
/* streaming protocol */
if (listen(sockfd, 5) < 0) {
iemnet_log(x, IEMNET_ERROR, "unable to listen on socket");
sys_sockerror("listen");
iemnet__closesocket(sockfd, 1);
sockfd = -1;
outlet_anything(x->x_statusout, gensym("port"), 1, ap);
return;
} else {
/* wait for new connections */
sys_addpollfn(sockfd,
(t_fdpollfn)tcpreceive_connectpoll,
x);
}
x->x_connectsocket = sockfd;
x->x_port = portno;
/* find out which port is actually used (useful when assigning "0") */
if(!getsockname(sockfd, (struct sockaddr *)&server, &serversize)) {
x->x_port = ntohs(server.sin_port);
}
SETFLOAT(ap, x->x_port);
outlet_anything(x->x_statusout, gensym("port"), 1, ap);
}
static void tcpreceive_serialize(t_tcpreceive *x, t_floatarg doit)
{
x->x_serialize = doit;
}
static void tcpreceive_free(t_tcpreceive *x)
{
/* is this ever called? */
if (x->x_connectsocket >= 0) {
sys_rmpollfn(x->x_connectsocket);
iemnet__closesocket(x->x_connectsocket, 1);
}
tcpreceive_disconnect_all(x);
if(x->x_floatlist) {
iemnet__floatlist_destroy(x->x_floatlist);
}
x->x_floatlist = NULL;
}
static void *tcpreceive_new(t_floatarg fportno)
{
t_tcpreceive*x;
int portno = fportno;
int i;
x = (t_tcpreceive *)pd_new(tcpreceive_class);
x->x_msgout = outlet_new(&x->x_obj, 0);
x->x_addrout = outlet_new(&x->x_obj, gensym("list")); /* legacy */
x->x_connectout = outlet_new(&x->x_obj, gensym("float")); /* legacy */
x->x_statusout = outlet_new(&x->x_obj, 0); /* outlet for everything else */
x->x_serialize = 1;
x->x_connectsocket = -1;
x->x_port = -1;
x->x_nconnections = 0;
/* clear the connection list */
for (i = 0; i < MAX_CONNECTIONS; ++i) {
x->x_connection[i].socket = -1;
x->x_connection[i].addr = 0L;
x->x_connection[i].port = 0;
}
x->x_floatlist = iemnet__floatlist_create(1024);
tcpreceive_port(x, portno);
return (x);
}
IEMNET_EXTERN void tcpreceive_setup(void)
{
if(!iemnet__register(objName)) {
return;
}
tcpreceive_class = class_new(gensym(objName),
(t_newmethod)tcpreceive_new, (t_method)tcpreceive_free,
sizeof(t_tcpreceive),
0,
A_DEFFLOAT, 0);
class_addmethod(tcpreceive_class, (t_method)tcpreceive_port,
gensym("port"), A_DEFFLOAT, 0);
class_addmethod(tcpreceive_class, (t_method)tcpreceive_serialize,
gensym("serialize"), A_FLOAT, 0);
DEBUGMETHOD(tcpreceive_class);
}
IEMNET_INITIALIZER(tcpreceive_setup);
/* end x_net_tcpreceive.c */