send data between two client sockets

Question

I have to make an app using C sockets on Mac-OS that sends data from one socket to other socket, like this.

1. Server waits for connections
2. Client connect to server(from 1). -> socket1
3. Server connects to an external server and obtains an socket. -> socket2

From now on the server job is finish. The data exchange should be made only between the client socket (from 2) and socket obtained from 3.

Current implementation: Server makes the connection and then reads data from one socket and sends to other.

Any ides how after step 3 to pipe line the two sockets socket1 and socket2.

Answer 1

Well your problem can be solved in two ways:

1) You need to code the part related to the connection formation between client and external server. But this puts an extra overload on the client, because it needs to make two connections, to both the servers (and I strongly feel the middle server in this case is useless).

2) Second way of solving it is passing the sockets between the servers: Client connects to the server, this middle server sends this socket to the external server. Now external server starts communication with the client. This can be done only if both the server processes run on the same machine. And the file-descriptors are usually passed using Unix Domain Sockets.

Here is the code which I have. You can use these two functions to either send or receive the file-descriptors. It works on my Linux machine. I don't know about Mac-OS.

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <sys/uio.h>
#include <unistd.h>
#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
/* this function passes 'fd_to_send' 
file descriptor via 
a unix domain socket 'sfd'...
*/
void pass_fd(int sfd, int fd_to_send)
{
    struct msghdr msg;

    /*allocate memory to 'msg_control' field in msghdr struct */
    char buf[CMSG_SPACE(sizeof(int))];
    /*the memory to be allocated should include data + header..
    this is calculated by the above macro...(it merely adds some
    no. of bytes and returs that number..*/

    struct cmsghdr *cmsg;

    struct iovec ve;    
    /*must send/receive atleast one byte...
    main purpose is to have some error 
    checking.. but this is completely 
    irrelevant in the current context..*/

    char *st ="I";
    /*jst let us allocate 1 byte for formality 
    and leave it that way...*/
    ve.iov_base = st;
    ve.iov_len =1;

    /*attach this memory to our main msghdr struct...*/
    msg.msg_iov = &ve;
    msg.msg_iovlen = 1;

    /*these are optional fields ..
    leave these fields with zeros..
    to prevent unnecessary SIGSEGVs..*/
    msg.msg_name = NULL;
    msg.msg_namelen = 0;


    /*here starts the main part..*/
    /*attach the 'buf' to msg_control..
    and fill in the size field correspondingly..
    */

    msg.msg_control = buf;
    msg.msg_controllen = sizeof(buf);

    /*actually msg_control field must 
    point to a struct of type 'cmsghdr'
    we just allocated the memory, yet we need to 
    set all the corresponding fields..
    It is done as follows:
    */
    cmsg = CMSG_FIRSTHDR(&msg);
    /* this macro returns the address in the buffer..
    from where the first header starts..
    */

    /*set all the fields appropriately..*/
    cmsg->cmsg_level = SOL_SOCKET;
    cmsg->cmsg_type = SCM_RIGHTS;
    cmsg->cmsg_len = CMSG_LEN(sizeof(fd_to_send));
    /*in the above field we need to store
    the size of header + data(in this case 4 bytes(int) for our fd..
    this is returned by the 'CMSG_LEN' macro..*/

    *(int*)CMSG_DATA(cmsg) = fd_to_send;
    /*after the above three fields we keep the actual data..
    the macro 'CMSG_DATA' returns pointer to this location
    and we set it to the file descriptor to be sent..
    */

    msg.msg_controllen = cmsg->cmsg_len;
    /*now that we have filled the 'cmsg' struct 
    we store the size of this struct..*/
    /*this one isn't required when you
    pass a single fd..
    but useful when u pass multiple fds.*/

    msg.msg_flags = 0;
    /*leave the flags field zeroed..*/

    if(sendmsg( sfd, &msg, 0)==-1){ perror("snd:\n"); exit(1); }
    /*send this over the UNIX deomain socoket..*/ 
    printf("sent fd:%d\n", fd_to_send);
    close(fd_to_send);
    /*close the fd which was sent..*/
}
/*returns the received fd over the unix domain socket 'sfd'..*/
int recv_fd(int sfd)
{
    struct msghdr msg;
    /*do all the unwanted things first...
    same as the send_fd function..*/
    struct iovec io;
    char ptr[1];
    io.iov_base = ptr;
    io.iov_len = 1;
    msg.msg_name = 0;
    msg.msg_namelen = 0;
    msg.msg_iov = &io;
    msg.msg_iovlen = 1;
    /*-----------------------*/


    char buf[CMSG_SPACE(sizeof(int))];
    msg.msg_control = buf;
    msg.msg_controllen = sizeof(buf);
    /*reasoning is same..as above*/

    /*now here comes the main part..*/

    if(recvmsg( sfd, &msg, 0)==-1)
    {
        /*some shit has happened*/
        perror("recv\n");
        exit(1);
    }

    struct cmsghdr *cm;

    cm =  CMSG_FIRSTHDR(&msg);
    /*get the first message header..*/

    if(cm->cmsg_type != SCM_RIGHTS)
    {
        /*again some shit has happened..*/
        perror("unknown type..\n");
        exit(1);
    }

    /*if control has reached here.. this means
    we have got the correct message..and when you 
    extract the fd out of this message 
    this need not be same as the one which was sent..
    allocating a new fd is all done by the kernel
    and our job is jst to use it..*/
     printf("received fd:%d\n", *(int*)CMSG_DATA(cm));
     return *(int*)CMSG_DATA(cm);
}               

Answer 2

In the below example:

ClientOne and ClientTwo connect to the server. When the server receives both ClientOne and ClientTwo's socket descriptor, it sends ClientOne's information to ClientTwo and vice-versa.

The information it sends is the IP and the client is coming from. Server shuts down.

When each client receives their info, a socket is created and they connect to eachother. The server socket is then shutdown.

Socket Class:

#include <winsock2.h>
#include <Ws2tcpip.h>
#include <windows.h>
#include <cstdint>
#include <string>
#include <stdexcept>
#include <iostream>
#include <thread>
#include <vector>

class Socket
{
    private:
        SOCKET socket;
        std::uint32_t Port;
        std::string Address;
        bool Listen, Initialized, Asynchronous;
        void Swap(Socket &S);
        void UnInitialized();

    public:
        Socket();
        Socket(std::uint32_t Port, std::string Address, bool Listen = false, bool Asynchronous = false);
        Socket(const Socket &S) = delete;
        Socket(Socket && S);
        ~Socket();

        Socket& operator = (const Socket &S) = delete;
        Socket& operator = (Socket && S);

        int Recv(void* Buffer, std::uint32_t BufferLength);
        int Recv(SOCKET S, void* Buffer, std::uint32_t BufferLength);
        std::uint32_t RecvEx(void* Buffer, std::uint32_t BufferLength);
        std::uint32_t RecvEx(SOCKET S, void* Buffer, std::uint32_t BufferLength);

        int Send(void* Buffer, std::size_t BufferSize);
        int Send(SOCKET S, void* Buffer, std::size_t BufferSize);
        void Connect();
        void Connect(std::uint32_t Port, std::string Address, bool Listen, bool Asynchronous);
        SOCKET Accept(sockaddr* ClientInfo, int* ClientInfoSize);
        void Close();

        SOCKET GetSocket() const;
};

Socket::~Socket()
{
    Close();
}

void Socket::Close()
{
    if (socket)
    {
        shutdown(socket, SD_BOTH);
        closesocket(socket);
        socket = 0;
    }

    if (Initialized)
    {
        WSACleanup();
    }
}

SOCKET Socket::GetSocket() const
{
    return this->socket;
}

Socket::Socket(Socket && S) : socket(std::move(S.socket)), Port(std::move(S.Port)), Address(std::move(S.Address)), Listen(std::move(S.Listen)), Initialized(std::move(S.Initialized)), Asynchronous(std::move(S.Asynchronous)) {}

Socket::Socket() : socket(0), Port(0), Address(std::string()), Listen(false), Initialized(false), Asynchronous(false) {}

Socket::Socket(std::uint32_t Port, std::string Address, bool Listen, bool Asynchronous) : socket(0), Port(Port), Address(Address), Listen(Listen), Initialized(true), Asynchronous(Asynchronous)
{
    Connect(Port, Address, Listen, Asynchronous);
}

void Socket::Connect()
{
    UnInitialized();
    Connect(Port, Address, Listen, Asynchronous);
}

void Socket::Connect(std::uint32_t Port, std::string Address, bool Listen, bool Asynchronous)
{
    if (!socket)
    {
        this->Port = Port;
        this->Address = Address;
        this->Asynchronous = Asynchronous;
        this->Initialized = true;

        WSADATA wsaData;
        struct sockaddr_in* sockaddr_ipv4;

        if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
        {
            throw std::runtime_error("Error: WSAStartup Failed");
        }

        if (Address != "INADDR_ANY")
        {
            if (Address.find("http://") != std::string::npos)
            {
                Address = Address.substr(7);
            }

            std::size_t Position = Address.find("/");
            if (Position != std::string::npos)
            {
                Address = Address.substr(0, Position);
            }

            struct addrinfo *it = nullptr, *result = nullptr;
            getaddrinfo(Address.c_str(), nullptr, nullptr, &result);
            for (it = result; it != nullptr; it = it->ai_next)
            {
                sockaddr_ipv4 = reinterpret_cast<sockaddr_in*>(it->ai_addr);
                Address = inet_ntoa(sockaddr_ipv4->sin_addr);
                if (Address != "0.0.0.0") break;
            }
            freeaddrinfo(result);
        }

        if ((this->socket = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) == INVALID_SOCKET)
        {
            this->Close();
            throw std::runtime_error("Error: Failed to create socket");
        }

        struct sockaddr_in SockAddr;
        memset(&SockAddr, 0, sizeof(SockAddr));
        SockAddr.sin_port = htons(Port);
        SockAddr.sin_family = AF_INET;
        SockAddr.sin_addr.s_addr = (Address == "INADDR_ANY" ? htonl(INADDR_ANY) : inet_addr(Address.c_str()));

        if (Listen && (bind(this->socket, reinterpret_cast<SOCKADDR*>(&SockAddr), sizeof(SockAddr)) == SOCKET_ERROR))
        {
            this->Close();
            throw std::runtime_error("Error: Socket binding failed");
        }

        if (Listen && (listen(this->socket, SOMAXCONN) == SOCKET_ERROR))
        {
            this->Close();
            throw std::runtime_error("Error: Socket Listening Failed");
        }

        if(!Listen && (connect(this->socket, reinterpret_cast<SOCKADDR*>(&SockAddr), sizeof(SockAddr)) == SOCKET_ERROR))
        {
            if(Asynchronous && WSAGetLastError() != WSAEWOULDBLOCK)
            {
                this->Close();
                throw std::runtime_error("Error: Socket Connection failed");
            }
            else if (!Asynchronous)
            {
                this->Close();
                throw std::runtime_error("Error: Socket Connection failed");
            }
        }
    }
}

SOCKET Socket::Accept(sockaddr* ClientInfo, int* ClientInfoSize)
{
    static int Size = sizeof(sockaddr);
    return accept(this->socket, ClientInfo, (ClientInfo && ClientInfoSize ? ClientInfoSize : &Size));
}

Socket& Socket::operator = (Socket && S)
{
    S.Swap(*this);
    return *this;
}

int Socket::Recv(void* Buffer, std::uint32_t BufferLength)
{
    return recv(this->socket, reinterpret_cast<char*>(Buffer), BufferLength, 0);
}

int Socket::Recv(SOCKET S, void* Buffer, std::uint32_t BufferLength)
{
    return recv(S, reinterpret_cast<char*>(Buffer), BufferLength, 0);
}

std::uint32_t Socket::RecvEx(void* Buffer, std::uint32_t BufferLength)
{
    return this->RecvEx(this->socket, Buffer, BufferLength);
}

std::uint32_t Socket::RecvEx(SOCKET S, void* Buffer, std::uint32_t BufferLength)
{
    UnInitialized();
    char* Pointer = reinterpret_cast<char*>(Buffer);
    std::uint32_t TotalRead = 0;

    while (BufferLength > 0)
    {
        int BytesRead = recv(S, Pointer, std::min(1024 * 1024, static_cast<int>(BufferLength)), 0);
        if (BytesRead < 0)
        {
            if ((BytesRead == SOCKET_ERROR) && (WSAGetLastError() == WSAEWOULDBLOCK))
                continue;

            throw std::runtime_error("Error! RecvEx: Failed To Read Bytes.");
        }

        if (BytesRead == 0) break;

        Pointer += BytesRead;
        BufferLength -= BytesRead;
        TotalRead += BytesRead;
    }

    return TotalRead;
}

int Socket::Send(void* Buffer, std::size_t BufferSize)
{
    return send(this->socket, reinterpret_cast<char*>(Buffer), BufferSize, 0);
}

int Socket::Send(SOCKET S, void* Buffer, std::size_t BufferSize)
{
    return send(S, reinterpret_cast<char*>(Buffer), BufferSize, 0);
}

void Socket::Swap(Socket &S)
{
    using std::swap;
    swap(socket, S.socket);
    swap(Port, S.Port);
    swap(Address, S.Address);
    swap(Listen, S.Listen);
    swap(Initialized, S.Initialized);
    swap(Asynchronous, S.Asynchronous);
}

void Socket::UnInitialized()
{
    if (!Initialized)
    {
        throw std::runtime_error("\nError! Socket Not Constructed!");
        std::cout << "Socket Not Constructed!\n";
        ExitProcess(0);
    }
}

Server.cpp:

#include "Sockets.hpp"

#define PORT 27015
#define ADDRESS INADDR_ANY
#define CLIENTCOUNT 2

typedef struct
{
    std::string ip;
    int port;
    SOCKET sock;
} ClientInfo;

template <typename T>
inline T ReadPointer(TCHAR* &Pointer)
{
    T Result = *(reinterpret_cast<T*>(Pointer));
    Pointer += sizeof(T) / sizeof(TCHAR);
    return Result;
}

template <typename T>
inline void WritePointer(TCHAR* &Pointer, const T& Value)
{
    *(reinterpret_cast<T*>(Pointer)) = Value;
    Pointer += sizeof(T) / sizeof(TCHAR);
}

bool SendClient(ClientInfo* client, ClientInfo* receiver)
{
    int datasize = sizeof(client->ip.size()) + client->ip.size() + sizeof(client->port);
    std::vector<char> buffer(datasize, 0);
    char* ptr = &buffer[0];

    WritePointer(ptr, client->ip.size());
    for (std::size_t i = 0; i < client->ip.size(); ++i)
        WritePointer(ptr, client->ip[i]);

    WritePointer(ptr, client->port);

    std::cout << "Sending: " << &buffer[0] << "\n";

    return send(receiver->sock, &buffer[0], datasize, 0) == datasize;
}

bool ReadClient(SOCKET sock, ClientInfo* client)
{
    std::size_t ip_size = 0;
    recv(sock, (char*) &ip_size, sizeof(client->ip.size()), 0);
    client->ip.resize(ip_size);

    recv(sock, &client->ip[0], ip_size, 0);
    recv(sock, (char*) &client->port, sizeof(int), 0);

    std::cout<<client->ip<<"\n";
    return true;
}

int main()
{
    Socket s;
    s.Connect(PORT, "localhost", true, false);
    char buffer[1024] = {0};
    std::vector<ClientInfo> clients;

    while(true)
    {
        if (clients.size() < CLIENTCOUNT)
        {
            sockaddr_in ClientAddressInfo = {0};
            SOCKET sock = s.Accept(reinterpret_cast<sockaddr*>(&ClientAddressInfo), nullptr);
            char* ip = inet_ntoa(ClientAddressInfo.sin_addr);
            int port = (int) ntohs(ClientAddressInfo.sin_port);
            ClientInfo info = {ip, port, sock};
            clients.push_back(info);

            std::cout << "Client Connected From: " << ip << " on port: " << port << "\n";
        }

        if (ReadAsync(s, buffer))
        {
            std::cout << "Connected\n";
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(100));

        if (clients.size() >= CLIENTCOUNT)
        {
            SendClient(&clients[0], &clients[1]);
            SendClient(&clients[1], &clients[0]);
            return 0;
        }
    }
}

Client.cpp:

#define PORT 27015
#define ADDRESS INADDR_ANY
#define CLIENTCOUNT 2

typedef struct
{
    std::string ip;
    int port;
    SOCKET sock;
} ClientInfo;

template <typename T>
inline T ReadPointer(TCHAR* &Pointer)
{
    T Result = *(reinterpret_cast<T*>(Pointer));
    Pointer += sizeof(T) / sizeof(TCHAR);
    return Result;
}

template <typename T>
inline void WritePointer(TCHAR* &Pointer, const T& Value)
{
    *(reinterpret_cast<T*>(Pointer)) = Value;
    Pointer += sizeof(T) / sizeof(TCHAR);
}

bool SendClient(ClientInfo* client, ClientInfo* receiver)
{
    int datasize = sizeof(client->ip.size()) + client->ip.size() + sizeof(client->port);
    std::vector<char> buffer(datasize, 0);
    char* ptr = &buffer[0];

    WritePointer(ptr, client->ip.size());
    for (std::size_t i = 0; i < client->ip.size(); ++i)
        WritePointer(ptr, client->ip[i]);

    WritePointer(ptr, client->port);

    std::cout << "Sending: " << &buffer[0] << "\n";

    return send(receiver->sock, &buffer[0], datasize, 0) == datasize;
}

bool ReadClient(SOCKET sock, ClientInfo* client)
{
    std::size_t ip_size = 0;
    recv(sock, (char*) &ip_size, sizeof(client->ip.size()), 0);
    client->ip.resize(ip_size);

    recv(sock, &client->ip[0], ip_size, 0);
    recv(sock, (char*) &client->port, sizeof(int), 0);
    return true;
}

bool ReadAsync(const Socket &sock, ClientInfo* client)
{
    struct timeval tv;
    tv.tv_sec = 0;
    tv.tv_usec = 100000;
    fd_set rfds;
    FD_ZERO(&rfds);
    FD_SET(sock.GetSocket(), &rfds);
    for (int i = 0; i < 600; ++i)
    {
        if (select(sock.GetSocket(), &rfds, &rfds, NULL, &tv))
        {
            return ReadClient(sock.GetSocket(), client);
        }
        tv.tv_sec = 0;
        tv.tv_usec = 100000;
    }
    return false;
}

int main()
{
    Socket s;
    s.Connect(PORT, "localhost", false, false);
    std::vector<SOCKET> clients;
    ClientInfo client = {};

    while(true)
    {
        if (ReadAsync(s, &client))
        {
            std::cout<<"IP: "<<client.ip<<" PORT: "<<client.port<<"\n";
            s = std::move(Socket(client.port, client.ip, true, false));
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        system("CLS");
        std::cout<<"Connecting..\n";
    }
}