crossthread.win.cc

Go to the documentation of this file.

/*
  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;
}