13
0
livetrax/libs/pbd/crossthread.win.cc

197 lines
5.2 KiB
C++

/*
Copyright (C) 2009 Paul Davis
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, write to the Free Software
Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
*/
CrossThreadChannel::CrossThreadChannel (bool non_blocking)
: receive_channel (0)
, receive_source (0)
, receive_slot ()
, send_socket()
, receive_socket()
, recv_address()
{
struct sockaddr_in send_address;
// Create Send Socket
send_socket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
send_address.sin_family = AF_INET;
send_address.sin_addr.s_addr = inet_addr("127.0.0.1");
send_address.sin_port = htons(0);
int status = bind(send_socket, (SOCKADDR*)&send_address,
sizeof(send_address));
if (status != 0) {
std::cerr << "CrossThreadChannel::CrossThreadChannel() Send socket binding failed with error: " << WSAGetLastError() << std::endl;
return;
}
// make the socket non-blockable if required
u_long mode = (u_long)non_blocking;
int otp_result = 0;
otp_result = ioctlsocket(send_socket, FIONBIO, &mode);
if (otp_result != NO_ERROR) {
std::cerr << "CrossThreadChannel::CrossThreadChannel() Send socket cannot be set to non blocking mode with error: " << WSAGetLastError() << std::endl;
}
// Create Receive Socket, this socket will be set to unblockable mode by IO channel
receive_socket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
recv_address.sin_family = AF_INET;
recv_address.sin_addr.s_addr = inet_addr("127.0.0.1");
recv_address.sin_port = htons(0);
status = bind(receive_socket, (SOCKADDR*)&recv_address,
sizeof(recv_address));
if (status != 0) {
std::cerr << "CrossThreadChannel::CrossThreadChannel() Receive socket binding failed with error: " << WSAGetLastError() << std::endl;
return;
}
// recieve socket will be made non-blocking by GSource which will use it
// get assigned port number for Receive Socket
int recv_addr_len = sizeof(recv_address);
status = getsockname(receive_socket, (SOCKADDR*)&recv_address, &recv_addr_len);
if (status != 0) {
std::cerr << "CrossThreadChannel::CrossThreadChannel() Setting receive socket address to local failed with error: " << WSAGetLastError() << std::endl;
return;
}
// construct IOChannel
receive_channel = g_io_channel_win32_new_socket((gint)receive_socket);
// set binary data type
GIOStatus g_status = g_io_channel_set_encoding (receive_channel, NULL, NULL);
if (G_IO_STATUS_NORMAL != g_status ) {
std::cerr << "CrossThreadChannel::CrossThreadChannel() Cannot set flag for IOChannel. " << g_status << std::endl;
return;
}
// disable channel buffering
g_io_channel_set_buffered (receive_channel, false);
}
CrossThreadChannel::~CrossThreadChannel ()
{
/* glibmm hack */
if (receive_channel) {
g_io_channel_unref (receive_channel);
}
closesocket(send_socket);
closesocket(receive_socket);
}
void
CrossThreadChannel::wakeup ()
{
char c = 0;
// write one byte to wake up a thread which is listening our IOS
sendto(send_socket, &c, sizeof(c), 0, (SOCKADDR*)&recv_address, sizeof(recv_address) );
}
void
CrossThreadChannel::drain ()
{
/* flush the buffer - empty the channel from all requests */
GError *g_error = 0;
gchar buffer[512];
gsize read = 0;
while (1) {
GIOStatus g_status = g_io_channel_read_chars (receive_channel, buffer, sizeof(buffer), &read, &g_error);
if (G_IO_STATUS_AGAIN == g_status) {
break;
}
if (G_IO_STATUS_NORMAL != g_status) {
std::cerr << "CrossThreadChannel::CrossThreadChannel() Cannot drain from read buffer! " << g_status << std::endl;
if (g_error) {
std::cerr << "Error is Domain: " << g_error->domain << " Code: " << g_error->code << std::endl;
g_clear_error(&g_error);
} else {
std::cerr << "No error provided\n";
}
break;
}
}
}
int
CrossThreadChannel::deliver (char msg)
{
// write one particular byte to wake up the thread which is listening our IOS
int status = sendto(send_socket, &msg, sizeof(msg), 0, (SOCKADDR*)&recv_address, sizeof(recv_address) );
if (SOCKET_ERROR == status) {
return -1;
}
return status;
}
bool
CrossThreadChannel::poll_for_request()
{
// windows before Vista has no poll
while(true) {
fd_set rfds;
FD_ZERO(&rfds);
FD_SET(receive_socket, &rfds);
if ((select(receive_socket+1, &rfds, NULL, NULL, NULL)) < 0) {
if (errno == EINTR) {
continue;
}
break;
}
if(FD_ISSET(receive_socket, &rfds)) {
return true;
}
}
return false;
}
int
CrossThreadChannel::receive (char& msg, bool wait)
{
gsize read = 0;
GError *g_error = 0;
if (wait) {
if (!poll_for_request ()) {
return -1;
}
}
// fetch the message from the channel.
GIOStatus g_status = g_io_channel_read_chars (receive_channel, &msg, sizeof(msg), &read, &g_error);
if (G_IO_STATUS_NORMAL != g_status) {
read = -1;
}
return read;
}