tcp.c

Linux下的飞鸽传书

	return rc;
      rc=create_response(IPMSG_FILE_RETPARENT,info->name,info->size,top_dir,&res_message);
      if (rc<0)
	return rc;
      dbg_out("Send ret parent:%s\n",res_message);
      rc=send_header(con,res_message);
      g_free(res_message);
      if (rc<0)
	return rc;
    }
    res=gnome_vfs_directory_read_next(handle,info);
    if (res!=GNOME_VFS_OK) {
      if (res != GNOME_VFS_ERROR_EOF)
	err_out("Can not read next dir:%s %s (%d)\n",
		top_dir,
		gnome_vfs_result_to_string(res),
		res);
      goto error_out;
    }
  }

 error_out:
  gnome_vfs_directory_close(handle); 
  g_free(uri);
  return rc;
}
static int
finalize_send_directory(tcp_con_t *con,const char *top_dir,GnomeVFSFileInfo *info){
  int rc;
  char *res_message;

  if ( (!top_dir) || (!info) )
    return -EINVAL;
  dbg_out("Here\n",res_message,top_dir);
  rc=create_response(IPMSG_FILE_RETPARENT,info->name,info->size,top_dir,&res_message);
  if (rc<0) {
    err_out("finalizesend directory fail %s (%d)\n",
	    strerror(-rc),rc);
    return rc;
  }
    
  dbg_out("Send return to top dir:%s (%s)\n",res_message,top_dir);
  rc=send_header(con,res_message);
  g_free(res_message);

  return rc;
}


static int
tcp_transfer_dir(tcp_con_t *con,const char *path){
  int rc;
  char *basename;
  GnomeVFSFileInfo *info;
  GnomeVFSResult res;
  gchar *dir_uri;

  if ( (!con) || (!path) )
    return -EINVAL;
  info=gnome_vfs_file_info_new();
  dir_uri=gnome_vfs_get_uri_from_local_path(path);

  rc=-ENOMEM;
  if (!dir_uri) 
    goto free_info_out;

  res=gnome_vfs_get_file_info(dir_uri,info,GNOME_VFS_FILE_INFO_FOLLOW_LINKS);
  if (res != GNOME_VFS_OK) {
    rc=-res;
    goto free_info_out;
  }
  basename=g_path_get_basename(path);
    rc=-ENOMEM;
  if (!basename)
    goto free_info_out;

  rc=send_directory(con,path,basename,info);
  g_free(basename);
  if (!rc)
    rc=finalize_send_directory(con,path,info);
  else
    err_out("Send directory fail %s (%d)\n",
	    strerror(-rc),rc);
 free_info_out:
  gnome_vfs_file_info_unref(info);
  return rc;
}
int
disable_pipe_signal(struct sigaction *saved_act){
  int rc;
  struct sigaction new_act;
  
  if (!saved_act)
    return -EINVAL;

  memset(&new_act,0,sizeof(struct sigaction));

  new_act.sa_handler=SIG_IGN;
  rc=sigaction(SIGPIPE,&new_act,saved_act);
  g_assert(!rc);
  
  return rc;
}
int
enable_pipe_signal(struct sigaction *saved_act){
  int rc;
  
  if (!saved_act)
    return -EINVAL;

  rc=sigaction(SIGPIPE,saved_act,NULL);
  g_assert(!rc);
  
  return rc;
  
}
int
tcp_enable_keepalive(const tcp_con_t *con){
  int rc;
  int flag;

  if (!con) 
    return -EINVAL;

  flag=1;
  rc=setsockopt(con->soc, SOL_SOCKET, SO_KEEPALIVE, (void*)&flag, sizeof(int));
  if (rc<0) {
    err_out("Can not set keepalive:%s(%d)\n",strerror(errno),errno);
    return -errno;
  }

  return 0;
}

int
tcp_setup_client(int family,const char *ipaddr,int port,tcp_con_t *con){
  int rc;
  int sock=-1;
  struct sockaddr_in server;
  struct hostent *host;
  struct addrinfo *info=NULL;

  if ( (!ipaddr) || (!con) )
    return -EINVAL;

  rc=setup_addr_info(&info, ipaddr,port,SOCK_STREAM,family);
  if (rc<0)
    return rc;

  sock=socket(info->ai_family, info->ai_socktype,info->ai_protocol);

  if (sock<0) {
    err_out("Can not create socket: %s(errno:%d)\n",strerror(errno),errno);
    rc=-errno;
    goto error_out;    
  }

#ifdef IPV6_V6ONLY
  if (info->ai_family == AF_INET6) {
    int v6only=1;

    rc=setsockopt(sock,IPPROTO_IPV6,IPV6_V6ONLY,(void *)&v6only,sizeof(v6only));
    if (rc) {
      goto error_out;
    }
  }
#endif /*  IPV6_V6ONLY  */

  if (connect(sock, info->ai_addr, info->ai_addrlen)<0){
    err_out("Can not connect addr:%s port:%d %s (errno:%d)\n",
	    ipaddr,
	    port,
	    strerror(errno),
	    errno);
    rc=-errno;
    goto error_out;    
  }
  rc=sock_set_buffer(sock);
  if (rc<0) {
    err_out("Can not set socket buffer: %s (%d)\n",strerror(errno),errno);
    return -errno;
  }
  /* 受信時はタイムアウト待ちを行うのでここでは, ノンブロックにしないこと  */
  con->soc=sock;
  con->family=info->ai_family;
  con->self=info;

  return 0;
 error_out:
  if (info)
    freeaddrinfo(info);
  return rc;
}

int
tcp_set_buffer(tcp_con_t *con) {
  if (!con)
    return -EINVAL;

 return sock_set_buffer(con->soc);
}
gpointer
ipmsg_tcp_recv_thread(gpointer data){
  ssize_t recv_len;
  size_t addr_len;
  int rc;
  tcp_con_t *con;
  char *recv_buf=NULL;
  int count=200;
  struct addrinfo *info;

  con=(tcp_con_t *)data;
  if (!con) {
    err_out("No connection recived\n");
    g_assert(con);
    return NULL;
  }
  if (!(con->self)) {
    err_out("Invalid connection recived\n");
    g_assert(con->self);
    return NULL;
  }
  
  rc=tcp_enable_keepalive(con);
  if (rc<0)
    return NULL;

  recv_buf=g_malloc(_MSG_BUF_SIZE);
  if (!recv_buf)
    return NULL;

  memset(recv_buf,0,_MSG_BUF_SIZE);

  while(1) {
    if (wait_socket(con->soc,WAIT_FOR_READ,TCP_SELECT_SEC)<0) {
      err_out("Can not send socket\n");
      destroy_tcp_connection(con);
      goto error_out;
    }else{
    read_retry:
      errno=0;
      memset(recv_buf, 0, sizeof(recv_buf));
      info=con->self;
      recv_len=recv(con->soc,
		    recv_buf,
		    _MSG_BUF_SIZE,
		    MSG_PEEK);

      if (recv_len<=0) {
	if (errno==EINTR)
	  goto read_retry;
	if (errno)
	  err_out("Can not peek message %s(errno:%d)\n",strerror(errno),errno);
	destroy_tcp_connection(con);
	goto error_out;
      }

      dbg_out("tcp peek read:%d %s\n",recv_len,recv_buf);

      recv_len=recv(con->soc,
		    recv_buf,
		    recv_len,
		    MSG_PEEK);

      dbg_out("tcp read:%d %s\n",recv_len,recv_buf);
      if (recv_len>0) {
	msg_data_t msg;
	request_msg_t req;
	char *path;
	size_t size;
	unsigned long ipmsg_fattr;

	recv_buf[recv_len-1]='\0';
	memset(&req,0,sizeof(request_msg_t));

	init_message_data(&msg);
	parse_message(NULL,&msg,recv_buf,recv_len);
	parse_request(&req,msg.message);
	rc=refer_attach_file(req.pkt_no,req.fileid,&ipmsg_fattr,(const char **)&path,&size);
	if (rc<0) {
	  err_out("Can not find message:pktno %ld id : %d\n",
		  req.pkt_no,
		  req.fileid);
	  close(con->soc); /* エラーとしてクローズする  */
	}else{
	  dbg_out("transfer:%s (%d)\n",path,size);
	  switch(ipmsg_fattr) 
	    {
	    case IPMSG_FILE_REGULAR:
	      if (!tcp_transfer_file(con,path,size,0)) {
		tcp_ipmsg_finalize_connection(con);
		download_monitor_release_file(req.pkt_no,req.fileid);
	      }
	      break;
	    case IPMSG_FILE_DIR:
	      if (!tcp_transfer_dir(con,path)){
		download_monitor_release_file(req.pkt_no,req.fileid);
		tcp_ipmsg_finalize_connection(con);
	      }
	      break;
	    default:
	      break;
	    }
	  release_message_data(&msg);
	  g_free(path);
	  break;
	}
	release_message_data(&msg);
	break;
      }
    }
    --count;
    if (!count)
      break;
  }
  if (con->self) /* closeは相手側で行うので, destroy_tcp_connectionは呼び出せない
		  * (Win版ipmsgの仕様).  
		  */
    freeaddrinfo(con->self); 
 error_out:
  g_free(con);
  if (recv_buf)
    g_free(recv_buf);

  return NULL;
}
gpointer 
ipmsg_tcp_server_thread(gpointer data){
  int rc;
  size_t len;
  int sock;
  int tcp_socket;
  int port;
  int reuse;
  int family;
  tcp_con_t *client_con=NULL;
  struct addrinfo *info=NULL;

  family=(int)data;
  rc=setup_addr_info(&info, 
		     NULL,
		     hostinfo_refer_ipmsg_port(),
		     SOCK_STREAM,family);
  if (rc<0)
    return NULL;

  tcp_socket = socket(info->ai_family,
		      info->ai_socktype,
		      info->ai_protocol);

  if (tcp_socket < 0) {
    err_out("Can not create socket:%s (%d)\n",strerror(errno),errno);
    goto error_out;
  }

  reuse=1;
  if (setsockopt(tcp_socket,SOL_SOCKET, SO_REUSEADDR, (const char *)&reuse, sizeof(reuse))<0) {
    err_out("Can not set sock opt:%s (%d)\n",strerror(errno),errno);
    goto error_out;
  }

#ifdef IPV6_V6ONLY
  if (info->ai_family == AF_INET6) {
    int v6only=1;

    rc=setsockopt(tcp_socket,IPPROTO_IPV6,IPV6_V6ONLY,(void *)&v6only,sizeof(v6only));
    if (rc) {
      goto error_out;
    }
  }
#endif /*  IPV6_V6ONLY  */

  if (bind(tcp_socket, info->ai_addr, info->ai_addrlen) != 0) {
    err_out("Can not bind socket:%s (%d)\n",strerror(errno),errno);
    goto error_out;
  }

  if (listen(tcp_socket, TCP_LISTEN_LEN) != 0) {
    err_out("Can not listen socket:%s (%d)\n",strerror(errno),errno);
    goto error_out;
  }

  if (sock_set_buffer(tcp_socket)<0) {
    err_out("Can not set socket buffer:%s (%d)\n",strerror(errno),errno);
    goto error_out;
  }

  while (1) {
    GThread *handler;
    struct addrinfo *client_info=NULL;

    client_con=g_malloc(sizeof(tcp_con_t));
    g_assert(client_con);
    dbg_out("accept\n");

    
    rc=setup_addr_info(&client_info, 
		     NULL,
		     hostinfo_refer_ipmsg_port(),
		     SOCK_STREAM,family);
    
    if (rc<0)
      goto free_client_out;
    
    sock = accept(tcp_socket, client_info->ai_addr,&(client_info->ai_addrlen));
    if (sock < 0) {
      err_out("Can not accept:%s (%d)\n",strerror(errno),errno);
      goto free_client_out;
    }
    if (sock_set_buffer(sock)<0) {
      err_out("Can not set socket buffer:%s (%d)\n",strerror(errno),errno);
      goto free_client_out;
    }

    client_con->soc=sock;
    client_con->self=client_info;
    handler=g_thread_create(ipmsg_tcp_recv_thread,
                            client_con,
			    FALSE,
			    NULL);
    if (!handler) {
      err_out("Can not create handler thread.\n");
      goto free_client_out;
    }
  }
  /* ここには, 来ないはずだが, 念のため  */
  if (info)
    freeaddrinfo(info);

  return NULL;

 free_client_out:  
  destroy_tcp_connection(client_con);
 error_out:
  if (info)
    freeaddrinfo(info);
  exit(0); /* 多重起動のため 終了  */
  return NULL;
}