(root)/iphone/common : /source/SocketMessenger.m (revision 100)

//

//  SocketMessenger.m

//  tappity

//

//  Created by döme on 30.10.2009.

//

/*

 * Copyright (c) 2009 Doemoetoer Gulyas.

 * All rights reserved.

 *

 * Redistribution and use in source and binary forms, with or without

 * modification, are permitted provided that the following conditions

 * are met:

 * 1. Redistributions of source code must retain the above copyright

 *    notice, this list of conditions and the following disclaimer.

 * 2. Redistributions in binary form must reproduce the above copyright

 *    notice, this list of conditions and the following disclaimer in the

 *    documentation and/or other materials provided with the distribution.

 * 3. The name of the author may not be used to endorse or promote products

 *    derived from this software without specific prior written permission.

 *

 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR

 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES

 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.

 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,

 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT

 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,

 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY

 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF

 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

 */

#include <unistd.h>

#include <stdint.h>

#include <fcntl.h>

#include <sys/types.h>

#include <sys/time.h>

#include <termios.h>

#include <netinet/in.h>

#include <sys/socket.h>

#include <arpa/inet.h>

#include <netdb.h>

#import "SocketMessenger.h"

NSString* SocketMessageIdKey	= @"SocketMessageId";

NSString* SocketMessageDataKey	= @"SocketMessageData";

NSString* SocketMessengerKey	= @"SocketMessenger";

//const int kSocketMessengerTerminateMsg = -1;

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));

}

@implementation SocketMessengerService

{

	int				listenSocket;

	NSNetService*	netService;

	uint16_t		servicePort;

	NSString*		serviceName;

	NSString*		protocolName;

	dispatch_queue_t					delegateQueue;

	dispatch_source_t	listenDispatchSource;

	SocketMessengerServiceErrorHandler errorHandler;

	SocketMessengerServiceConnectionCallback connectionCallback;

}

@synthesize errorHandler, connectionCallback;

- (id) init

{

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

		return nil;

	listenSocket = -1;

	self.delegateQueue = dispatch_get_main_queue();

	return self;

}

- (void) dealloc

{

	[self stopServiceSync];

}

- (void) finalize

{

	[super finalize];

}

- (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] && (listenSocket != -1));

	NSLog(@"Bonjour port: %d", servicePort);

	[netService stop];

	netService = [[NSNetService alloc] initWithDomain: domain type: protocol name: name port: servicePort];

	if(netService == nil)

		return NO;

	[netService setDelegate: self];

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

	[netService publish];

	return YES;

}

- (void) acceptConnection

{

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

	// accept

	socklen_t	sinSize = sizeof(struct sockaddr_in);

	struct sockaddr_in	peerAddress;

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

	_setStandardSocketOpts(newSocket);

	void (^errorBlock)() = ^{

		dispatch_source_cancel(listenDispatchSource);

		if (errorHandler)

			dispatch_async(delegateQueue, ^(void) {

				errorHandler([NSError errorWithDomain: @"SocketMessenger.accept" code: -1 userInfo: nil], self);

			});

	};

	if (newSocket == -1)

	{

		if (errno == EWOULDBLOCK)

		{

		}

		else

		{

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

			errorBlock();

			return;

		}

	}

	SocketMessenger* messenger = [[SocketMessenger alloc] init];

	if (connectionCallback)

	{

		NSLog(@"Dispatching connection callback...");

		dispatch_async(delegateQueue, ^(void) {

			connectionCallback(messenger, self);

		});

	}

	NSLog(@"Starting messenger");

	[messenger startWithSocket: newSocket];

	NSLog(@"Accepted connection.");

	return;

}

- (void) startService

{

	struct sockaddr_in myAddress;

	memset(&myAddress, 0, sizeof(myAddress));

	myAddress.sin_family = AF_INET;					// host byte order

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

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

	listenSocket = socket(PF_INET, SOCK_STREAM, 0);

	assert(listenSocket != -1);

	_setStandardSocketOpts(listenSocket);

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

	assert(-1 != err);

	err = listen(listenSocket, 1);

	assert(-1 != err);

	socklen_t slen = sizeof(myAddress);

	err = getsockname(listenSocket, (struct sockaddr *)&myAddress, &slen);

	assert(-1 != err);

	servicePort = ntohs(myAddress.sin_port);

	listenDispatchSource = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, listenSocket, 0, dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0));

	dispatch_source_set_cancel_handler(listenDispatchSource, ^{

		close(listenSocket);

		listenSocket = -1;

		dispatch_release(listenDispatchSource);

		listenDispatchSource = NULL;

	});

	dispatch_source_set_event_handler(listenDispatchSource, ^{ [self acceptConnection]; });

	dispatch_resume(listenDispatchSource);

	BOOL bonjourPartial = protocolName || serviceName;

	BOOL bonjourComplete = protocolName && serviceName;

	if (bonjourPartial && !bonjourComplete)

		NSLog(@"Bonjour service information incomplete, attempting to start anyway.");

	if (bonjourPartial)

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

			NSLog(@"socket messenger servie bounjour advertisments up and running");

		else

			NSLog(@"Bonjour service startup failed");

}

- (void) stopServiceSync

{

	[netService stop];

	netService = nil;

	if (listenDispatchSource)

		dispatch_source_cancel(listenDispatchSource);

	while (listenDispatchSource) usleep(1);

}

- (void) setDelegateQueue: (dispatch_queue_t) aQueue

{

	dispatch_retain(aQueue);

	if (delegateQueue)

		dispatch_release(delegateQueue);

	delegateQueue = aQueue;

}

- (BOOL) isListening

{

	return listenSocket != -1;

}

@synthesize protocolName, serviceName, servicePort;

@end

@interface SocketMessenger (Private)

- (void) startCommsWorkThreads;

- (void) startWatchdogThread;

- (void) startListening;

@end

@implementation SocketMessenger

{

	int commsSocket;

	dispatch_io_t		socketChannel;

	dispatch_source_t	watchdogDispatchSource;

	dispatch_source_t	servicePollDispatchSource;

	dispatch_queue_t	readQueue;

	dispatch_queue_t	writeQueue;

	dispatch_queue_t	watchdogQueue;

	dispatch_queue_t	delegateQueue;

	BOOL		automaticallyReconnect;

	BOOL		errorOccured;

	NSNetService* remoteService;

	SocketMessengerPacketReadHandler readHandler;

	SocketMessengerConnectionBeginCallback connectionBeginCallback;

	SocketMessengerConnectionEndCallback connectionEndCallback;

}

- (id) init

{

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

		return nil;

	commsSocket = -1;

	automaticallyReconnect = YES;

	self.readQueue = dispatch_get_main_queue();

	self.delegateQueue = dispatch_get_main_queue();

	self.writeQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);

	self.watchdogQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_LOW, 0);

	return self;

}

- (BOOL) isConnected

{

	return commsSocket != -1;

}

- (void) sendPlist: (id) plist withIdentifier: (NSInteger) messageId

{

	NSData* data = [NSPropertyListSerialization dataFromPropertyList: plist format: NSPropertyListBinaryFormat_v1_0 errorDescription: nil];

	assert(data);

	dispatch_data_t dispatchData = dispatch_data_create([data bytes], [data length], dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), DISPATCH_DATA_DESTRUCTOR_DEFAULT);

	[self sendPacket: dispatchData withIdentifier: messageId];

	dispatch_release(dispatchData);

}

- (void) sendPacket: (dispatch_data_t) payload withIdentifier: (NSInteger) messageId

{

	size_t payloadSize = dispatch_data_get_size(payload);

	uint32_t header[2] = {CFSwapInt32HostToBig(payloadSize+4), CFSwapInt32HostToBig(messageId)};

	dispatch_data_t headerData = dispatch_data_create(header, 8, NULL, nil);

	dispatch_data_t data = dispatch_data_create_concat(headerData, payload);

	if (socketChannel)

	{

		dispatch_io_write(socketChannel, 0, data, writeQueue, ^(bool done, dispatch_data_t data, int error){

			if(error)

			{

				NSLog(@"write error %d", error);

				errorOccured = YES;

			}

		});

	}

	dispatch_release(headerData);

	dispatch_release(data);

}

- (void) startWithSocket: (int) socket

{

	assert(commsSocket == -1);

	commsSocket = socket;

	[self startCommsWorkThreads];

}

- (void) connectToRemoteService

{

	NSInteger	port = [remoteService port];

	NSString*	hostName = [remoteService hostName];

	NSLog(@"SocketMessenger attempting to connect to %@ : %d", hostName, (int)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

	assert(sockfd != -1);

	NSLog(@"SocketMessenger connected");

	self->commsSocket = sockfd;

	[self startCommsWorkThreads];

	[self startWatchdogThread];

}

- (void) connectToService: (NSNetService*) service

{

	remoteService = service;

	[self connectToRemoteService];

}

- (void) stopCommsWorkThreads

{

	if (socketChannel)

	{

		dispatch_io_t channel = socketChannel;

		socketChannel = NULL;

		dispatch_io_close(channel, DISPATCH_IO_STOP);

		dispatch_release(channel);

	}

}

- (void) stopWatchdogThread

{

	if (watchdogDispatchSource)

		dispatch_source_cancel(watchdogDispatchSource);

	while (watchdogDispatchSource)

		usleep(1);

}

- (void) startWatchdogThread

{

	watchdogDispatchSource = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, watchdogQueue);

	dispatch_source_set_timer(watchdogDispatchSource, dispatch_walltime(DISPATCH_TIME_NOW, 0), 1ull*1000000000ull, 0ull); // 1s

	dispatch_source_set_event_handler(watchdogDispatchSource, ^(void) {

		if (errorOccured || (commsSocket == -1))

		{

			[self stopCommsWorkThreads];

			NSLog(@"watchdog noticed connection error");

			if (automaticallyReconnect)

			{

				NSLog(@"restarting client");

				[self connectToRemoteService];

			}

			else 

			{

				if (connectionEndCallback)

					dispatch_async(delegateQueue, ^{ connectionEndCallback(self); });

				dispatch_source_cancel(watchdogDispatchSource);

			}

		}

	});

	dispatch_source_set_cancel_handler(watchdogDispatchSource,  ^{

		dispatch_release(watchdogDispatchSource);

		watchdogDispatchSource = NULL;

	});

	dispatch_resume(watchdogDispatchSource);

}

- (void) startCommsWorkThreads

{

	[self stopCommsWorkThreads];

	errorOccured = NO;

	__block dispatch_data_t incompleteData = dispatch_data_empty;

	__block uint32_t amountToRead = 0;

	dispatch_retain(incompleteData);

	socketChannel = dispatch_io_create(DISPATCH_IO_STREAM, commsSocket, readQueue, ^(int error) {

		close(commsSocket);

		commsSocket = -1;

		dispatch_release(incompleteData);

		incompleteData = NULL;

	});

	dispatch_io_set_low_water(socketChannel, 0);

	dispatch_io_set_high_water(socketChannel, SIZE_MAX);

	__block void (^payloadReader)(_Bool done, dispatch_data_t newData, int error);

	__block void (^headerReader)(_Bool done, dispatch_data_t newData, int error) = ^(_Bool done, dispatch_data_t newData, int error)

	{

		assert(self);

		dispatch_data_t data = dispatch_data_create_concat(incompleteData, newData);

		size_t dataSize = dispatch_data_get_size(data);

		if (done && dataSize)

		{

			assert(dataSize == 4);

			const void* headerBuf = NULL;

			dispatch_data_t headerMap = dispatch_data_create_map(data, &headerBuf, NULL);

			memcpy(&amountToRead, headerBuf, 4);

			amountToRead = CFSwapInt32BigToHost(amountToRead);

			dispatch_release(incompleteData);

			incompleteData = dispatch_data_empty;

			dispatch_retain(incompleteData);

			dispatch_release(headerMap);

			if (amountToRead)

			{

				dispatch_io_read(socketChannel, 0, amountToRead, readQueue, payloadReader);

			}

			else if (!error && newData)

			{

				dispatch_io_read(socketChannel, 0, 4, readQueue, headerReader);

			}

			if (error)

				NSLog(@"oops, error while reading header: %i", error);

		}

		else

		{

			dispatch_release(incompleteData);

			incompleteData = data;

			dispatch_retain(incompleteData);

		}

		dispatch_release(data);

	};

	payloadReader = ^(_Bool done, dispatch_data_t newData, int error) {

		dispatch_data_t data = dispatch_data_create_concat(incompleteData, newData);

		size_t dataSize = dispatch_data_get_size(data);

		if (done)

		{

			assert(dataSize == amountToRead);

			amountToRead = 0;

			assert(dataSize >= 4);

			dispatch_data_t packetIdData = dispatch_data_create_subrange(data, 0, 4);

			dispatch_data_t payloadData = dispatch_data_create_subrange(data, 4, dataSize-4);

			const void* headerBuf = NULL;

			uint32_t packetIdentifier = 0;