(root)/iphone/tappity : 52 : common/source/SocketMessenger.m

//

//  SocketMessenger.m

//  tappity

//

//  Created by döme on 30.10.2009.

//  Copyright 2009 __MyCompanyName__. All rights reserved.

//

 

#include <unistd.h>

#include <stdint.h>

#include <sys/socket.h>

#include <fcntl.h>

#include <termios.h>

#include <netinet/in.h>

#include <arpa/inet.h>

#include <sys/types.h>

#include <netdb.h>

#include <netinet/in.h>

#include <sys/time.h>

 

#import "SocketMessenger.h"

 

NSString* SocketMessageIdKey    = @"SocketMessageId";

NSString* SocketMessageDataKey  = @"SocketMessageData";

 

@interface SocketMessenger (Private)

- (void) startCommsWorkThreads;

@end

 

@implementation SocketMessenger

 

- (id) init

{

        if (!(self = [super init]))

                return nil;

                

        commsSocket = -1;

        

        sendLock = [[NSCondition alloc] init];

                

        return self;

}

 

- (BOOL) threadActive: (id) key

{

        return [[activeThreads objectForKey: key] boolValue];

}

 

- (void) threadWillExit: (id) key

{

        @synchronized (self)

        {

                [activeThreads removeObjectForKey: key];

                

                if (![activeThreads count])

                {

                        close(commsSocket);

                        commsSocket = -1;

                        

                        

                }

        }

}

 

- (void) receivingThread: (id) info

{

        NSAutoreleasePool* pool = [[NSAutoreleasePool alloc] init];

 

        @synchronized(self)

        {

                [self retain];

        }

 

        while ([self threadActive: info])

        {

                struct timeval tv;

                fd_set readfds;

                fd_set writefds;

                fd_set errorfds;

                int socket = -1;

                int maxSocket = -1;

                

                @synchronized(self)

                {

                        socket = commsSocket;

                }

 

                tv.tv_sec = 1;

                tv.tv_usec = 0;

 

                FD_ZERO(&readfds);

                FD_ZERO(&writefds);

                FD_ZERO(&errorfds);

 

                FD_SET(socket, &readfds);

                FD_SET(socket, &errorfds);

                maxSocket = MAX(maxSocket, socket);

                

                if (maxSocket < 0)

                {

                        //printf("No sockets, sleeping for a bit...\n");

                        usleep(1000000);

                        continue;

                }

 

                if (select(maxSocket+1, &readfds, &writefds, &errorfds, &tv) < 0)

                {

                        perror("select");

                        break;

                }

 

                if (FD_ISSET(socket, &readfds))

                {

                        BOOL messageFinished = NO;

                        //NSLog(@"receiving...");

                        if (!expectedMessageSize)

                        {

                                uint32_t header[2] = {0,0};

                                int actuallyRead = 0;

                        

                                actuallyRead = recv(socket, header, 8, MSG_PEEK);

                                

                                if (actuallyRead == 8)

                                {

                                        actuallyRead = recv(socket, header, 8, 0);

                                        expectedMessageSize = ntohl(header[1]);

                                        currentMessageId = ntohl(header[0]);

                                        currentData = [[NSMutableData alloc] initWithLength: expectedMessageSize];

                                }

                                else if (actuallyRead == -1)

                                {

                                        //close(commsSocket);

                                        //self->commsSocket = 0;

                                        if (errno != ETIMEDOUT)

                                        {

                                                printf("Connection dropped with error.\n");

                                                break;

                                        }

                                }

                                else if (actuallyRead == 0)

                                {

                                        printf("remote socket closed.\n");

                                        break;

                                }

                                

                                if (!expectedMessageSize)

                                        messageFinished = YES;

                        }

                        else

                        {

                                size_t readAmount = expectedMessageSize - currentlyRead;

                                int actuallyRead = 0;

                        

                                actuallyRead = recv(socket, [currentData mutableBytes] + currentlyRead, readAmount, 0);

                                if (actuallyRead == -1)

                                {

                                        printf("Connection dropped with error.\n");

                                        break;

                                }

                                else if (actuallyRead == 0)

                                {

                                        printf("remote socket closed.\n");

                                        break;

                                }

 

                                currentlyRead += actuallyRead;

                                

                                if (currentlyRead == expectedMessageSize)

                                        messageFinished = YES;

                        }

 

                        if (messageFinished)

                        {

                                @synchronized(self)

                                {

                                        expectedMessageSize = 0;

                                        /*

                                        if (!receivedPackets)

                                                receivedPackets = [[NSMutableArray alloc] init];

                                        [receivedPackets addObject: currentData];

                                        [currentData release];

                                        */

                                        

                                }

 

                                if (receiveDataOnMainThread)

                                        [self performSelectorOnMainThread: @selector(dataReceived:) withObject: [NSDictionary dictionaryWithObjectsAndKeys: [NSNumber numberWithInt: currentMessageId], SocketMessageIdKey, currentData, SocketMessageDataKey, nil] waitUntilDone: NO];

                                else

                                        [self dataReceived: [NSDictionary dictionaryWithObjectsAndKeys: [NSNumber numberWithInt: currentMessageId], SocketMessageIdKey, currentData, SocketMessageDataKey, nil]];

 

                                currentData = nil;

                                currentlyRead = 0;

                        }

 

                }

        }

        

        [self threadWillExit: info];

 

        @synchronized(self)

        {

                [self release];

        }

 

        [info release];

        [pool drain];

}

 

 

- (void) runThreadWithSelector: (SEL) selector

{

        @synchronized(self)

        {

                if (!activeThreads)

                        activeThreads = [[NSMutableDictionary alloc] init];

                id number = [NSNumber numberWithInt: threadIds++];

 

                [activeThreads setObject: [NSNumber numberWithBool: YES] forKey: number];

                [NSThread detachNewThreadSelector: selector toTarget: self withObject: [number retain]];

        }

}

 

- (void) sendData: (NSData*) tapData withIdentifier: (NSInteger) messageId

{               

        [sendLock lock];

 

        if (!sendQueue)

                sendQueue = [[NSMutableArray alloc] init];

 

        [sendQueue addObject: [NSDictionary dictionaryWithObjectsAndKeys: [NSNumber numberWithInteger: messageId], SocketMessageIdKey, tapData, SocketMessageDataKey, nil]];

        

        [sendLock signal];

        [sendLock unlock];

                

}

 

 

- (void) sendingThread: (id) info

{

        NSAutoreleasePool* pool = [[NSAutoreleasePool alloc] init];

 

        @synchronized(self)

        {

                [self retain];

        }

 

        while ([self threadActive: info])

        {

                [sendLock lock];

                        while (![sendQueue count])

                                [sendLock wait];

                                

                //NSLog(@"sending");

                

                NSDictionary* dict = [[sendQueue objectAtIndex: 0] retain];

                NSData* data = [dict objectForKey: SocketMessageDataKey];

                uint32_t messageId = [[dict objectForKey: SocketMessageIdKey] intValue];

                [sendQueue removeObjectAtIndex: 0];

                

                [sendLock unlock];

 

                size_t sizeToSend = 8;

                size_t dataSent = 0;

                uint32_t header[2] = {htonl(messageId), htonl([data length])};

                int err = 0;

                int socket = -1;

 

                @synchronized(self)

                {

                        socket = commsSocket;

                }

 

                while (dataSent < sizeToSend)

                {

                        if ((err = send(socket, header + dataSent, sizeToSend - dataSent, 0)) == -1)

                        {

                                perror("send");

                                break;

                        }

                        else

                                dataSent += err;

                }

                

                if (err != -1)

                {

                        sizeToSend = [data length];

                        dataSent = 0;

                        while (dataSent < sizeToSend)

                        {

                                if ((err = send(socket, [data bytes] + dataSent, sizeToSend - dataSent, 0)) == -1)

                                {

                                        perror("send");

                                        break;

                                }

                                else

                                        dataSent += err;

                                        

                        }

                }

                [dict release];

                

                if (err == -1)

                        break;

                

        }

        

 

        [self threadWillExit: info];

 

        @synchronized(self)

        {

                [self release];

        }

 

        [info release];

        [pool drain];

}

 

- (void) connectToService: (NSNetService*) service

{

        NSInteger       port = [service port];

        NSString*       hostName = [service hostName];

        

        NSLog(@"SocketMessenger attempting to connect to %@ : %d", hostName, port);

 

        int sockfd = 0;

 

 

//      _setStandardSocketOpts(socket);

 

        struct addrinfo hints, *servinfo = NULL, *p = NULL;

        int rv = 0;

 

        memset(&hints, 0, sizeof hints);

        hints.ai_family = AF_UNSPEC; // use AF_INET6 to force IPv6

        hints.ai_socktype = SOCK_STREAM;

        //hints.ai_protocol = IPPROTO_TCP;

        hints.ai_flags    = AI_PASSIVE;

        

//      NSHost* host = [NSHost hostWithName: hostName];

 

        if ((rv = getaddrinfo([hostName UTF8String], [[NSString stringWithFormat: @"%d", port] UTF8String], &hints, &servinfo)) != 0)

        {

                fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(rv));

                exit(1);

        }

 

        // loop through all the results and connect to the first we can

        for(p = servinfo; p != NULL; p = p->ai_next)

        {

                if ((sockfd = socket(p->ai_family, p->ai_socktype, p->ai_protocol)) == -1)

                {

                        perror("socket");

                        continue;

                }

 

                if (connect(sockfd, p->ai_addr, p->ai_addrlen) == -1)

                {

                        close(sockfd);

                        perror("connect");

                        continue;

                }

 

                break; // if we get here, we must have connected successfully

        }

 

        if (p == NULL) {

                // looped off the end of the list with no connection

                fprintf(stderr, "failed to connect\n");

                exit(2);

        }

 

        freeaddrinfo(servinfo); // all done with this structure

 

        NSLog(@"SocketMessenger connected");

 

 

        self->commsSocket = sockfd;

        

        [self startCommsWorkThreads];

                

}

 

- (void) startCommsWorkThreads

{

        [self runThreadWithSelector: @selector(sendingThread:)];

        [self runThreadWithSelector: @selector(receivingThread:)];

        

        if ([self respondsToSelector: @selector(connectionWasEstablished)])

                [self connectionWasEstablished];

}

 

static void _setStandardSocketOpts(int socket)

{

    int yes = 1;

    setsockopt(socket, SOL_SOCKET, SO_REUSEADDR, (void *)&yes, sizeof(yes));

        int timeout = 2000;

        setsockopt(socket, SOL_SOCKET, SO_SNDTIMEO, (char*)&timeout, sizeof(timeout));

        setsockopt(socket, SOL_SOCKET, SO_RCVTIMEO, (char*)&timeout, sizeof(timeout));

}

 

 

- (void) acceptThread: (id) info

{

        NSAutoreleasePool* pool = [[NSAutoreleasePool alloc] init];

 

        @synchronized(self)

        {

                [self retain];

        }

 

        while ([self threadActive: info])

        {

 

                struct timeval tv;

                fd_set readfds;

                fd_set writefds;

                fd_set errorfds;

                int maxSocket = -1;

                

                int lsock = server.listenSocket;

 

                tv.tv_sec = 1;

                tv.tv_usec = 0;

 

                FD_ZERO(&readfds);

                FD_ZERO(&writefds);

                FD_ZERO(&errorfds);

 

                FD_SET(lsock, &readfds);

                FD_SET(lsock, &errorfds);

                maxSocket = MAX(maxSocket, lsock);

                

                if (maxSocket < 0)

                {

                        printf("No sockets, sleeping for a bit...\n");

                        usleep(1000000);

                        continue;

                }

                

                //NSLog(@"listening for connection...");

 

                if (select(maxSocket+1, &readfds, NULL, &errorfds, &tv) < 0)

                {

                        perror("select");

                        break;

                }

 

                if (FD_ISSET(lsock, &readfds))

                {

                        NSLog(@"Accepting connection...");

                        // accept

                        socklen_t       sinSize = sizeof(struct sockaddr_in);

                        struct sockaddr_in      peerAddress;

                        int newSocket = accept(lsock, (struct sockaddr *)&peerAddress, &sinSize);

                        

                        _setStandardSocketOpts(newSocket);

 

                        

                        if (newSocket == -1)

                        {

                                if (errno == EWOULDBLOCK)

                                {

                                }

                                else

                                {

                                        printf("Error accepting connection.\n");

                                        break;

                                }

                        }

                        

                        @synchronized(self)

                        {

                                if (commsSocket)

                                        close(commsSocket);

 

                                // add socket to sockets list

                                commsSocket = newSocket;

                        }

                        

                        [server.netService stop];

                        [server.netService release];

                        server.netService = nil;

                        

                        [self startCommsWorkThreads];

                        

                        NSLog(@"Accepted connection.");

                }

 

        }

 

        @synchronized(self)

        {

                [self release];

        }

 

        [info release];

        [pool drain];

}

 

 

- (BOOL) enableBonjourWithDomain:(NSString*)domain applicationProtocol:(NSString*)protocol name:(NSString*)name 

{

        if(![domain length])

                domain = @""; //Will use default Bonjour registration doamins, typically just ".local"

        if(![name length])

                name = @""; //Will use default Bonjour name, e.g. the name assigned to the device in iTunes

        

        assert([protocol length] && server.listenSocket);

        

        NSLog(@"tappity port: %d", ntohs(server.myAddress.sin_port));

 

        [server.netService stop];

        [server.netService release];

 

        server.netService = [[NSNetService alloc] initWithDomain: domain type: protocol name: name port: ntohs(server.myAddress.sin_port)];

        if(server.netService == nil)

                return NO;

        

        [server.netService setDelegate:self];

//      [server.netService scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSRunLoopCommonModes];

        [server.netService publish];

        

        return YES;

}

 

 

- (void) startListening

{

        server.myAddress.sin_family = AF_INET;                                  // host byte order

        server.myAddress.sin_port = htons(server.portnum);              // short, network byte order, any port

        server.myAddress.sin_addr.s_addr = htonl(INADDR_ANY);   // auto-fill with my IP

 

        server.listenSocket = socket(PF_INET, SOCK_STREAM, 0);

        assert(server.listenSocket != -1);

 

        _setStandardSocketOpts(server.listenSocket);

 

        int err = bind(server.listenSocket, (struct sockaddr *)&server.myAddress, sizeof(server.myAddress));

        assert(-1 != err);

        

        err = listen(server.listenSocket, 1);

        assert(-1 != err);

 

        [self runThreadWithSelector: @selector(acceptThread:)];

        

        if([self enableBonjourWithDomain: @"" applicationProtocol: server.protocolName name: server.serviceName])

                NSLog(@"tappity bounjour advertisments up and running");

 

}

 

- (void) startBonjourServerWithName: (NSString*) name protocol: (NSString*) protocol port: (int) pnum;

{

        NSLog(@"starting tappity server");

        

        [server.serviceName release];

        server.serviceName = [name retain];

        

        

        [server.protocolName release];

        server.protocolName = [protocol retain];

        

        server.portnum = pnum;

        

        [self startListening];

 

}

 

- (void) dealloc

{

        [server.serviceName release];

 

        [server.netService stop];

        [server.netService release];

 

        [activeThreads release];

        [sendQueue release];

        [sendLock release];

 

        [super dealloc];

}

 

 

@end