//// PGTSAsynchronousConnector.m// BaseTen//// Copyright (C) 2008-2010 Marko Karppinen & Co. LLC.//// Before using this software, please review the available licensing options// by visiting http://www.karppinen.fi/baseten/licensing/ or by contacting// us at sales@karppinen.fi. Without an additional license, this software// may be distributed only in compliance with the GNU General Public License.////// This program is free software; you can redistribute it and/or modify// it under the terms of the GNU General Public License, version 2.0,// as published by the Free Software Foundation.//// 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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA//// $Id$//#import "PGTSAsynchronousConnector.h"#import "PGTSConstants.h"#import "PGTSConnection.h"#import "BXError.h"#import "BXLogger.h"#import "BXEnumerate.h"#import "BXArraySize.h"#import <netdb.h>#import <arpa/inet.h>@implementationPGTSAsynchronousConnectorstaticvoidSocketReady(CFSocketRefs,CFSocketCallBackTypecallBackType,CFDataRefaddress,constvoid*data,void*self){[(id)selfsocketReady:callBackType];}-(id)init{if((self=[superinit])){mHostResolver=[[BXHostResolveralloc]init];[mHostResolversetDelegate:self];mExpectedCallBack=0;[selfsetCFRunLoop:CFRunLoopGetCurrent()];}returnself;}-(CFRunLoopRef)CFRunLoop{returnmRunLoop;}-(void)setCFRunLoop:(CFRunLoopRef)aRef{if(mRunLoop!=aRef){if(mRunLoop)CFRelease(mRunLoop);if(aRef){mRunLoop=aRef;CFRetain(mRunLoop);}}}-(void)setConnectionDictionary:(NSDictionary*)aDict{if(mConnectionDictionary!=aDict){[mConnectionDictionaryrelease];mConnectionDictionary=[aDictretain];}}-(void)freeCFTypes{//Don't release the connection. Delegate will handle it.if(mSocketSource){CFRunLoopSourceInvalidate(mSocketSource);CFRelease(mSocketSource);mSocketSource=NULL;}if(mSocket){CFSocketInvalidate(mSocket);CFRelease(mSocket);mSocket=NULL;}if(mRunLoop){CFRelease(mRunLoop);mRunLoop=NULL;}}-(void)cancel{[mHostResolvercancelResolution];if(mConnection){PQfinish(mConnection);mConnection=NULL;}}-(void)dealloc{[selffreeCFTypes];[selfcancel];[mConnectionErrorrelease];[superdealloc];}-(void)finalize{[selffreeCFTypes];[selfcancel];[superfinalize];}-(void)prepareForConnect{[superprepareForConnect];bzero(&mHostError,sizeof(mHostError));}#pragma mark Callbacks-(void)hostResolverDidSucceed:(BXHostResolver*)resolveraddresses:(NSArray*)addresses{BOOLreachedServer=NO;if(addresses){charaddressBuffer[40]={};// 8 x 4 (hex digits in IPv6 address) + 7 (colons) + 1 (nul character)NSMutableDictionary*connectionDictionary=[[mConnectionDictionarymutableCopy]autorelease];[connectionDictionaryremoveObjectForKey:kPGTSHostKey];//This is safe because each address is owned by the addresses CFArray which is owned //by mHost which is CFRetained.BXEnumerate(addressData,e,[addressesobjectEnumerator]){conststructsockaddr*address=[addressDatabytes];sa_family_tfamily=address->sa_family;void*addressBytes=NULL;switch(family){caseAF_INET:addressBytes=&((structsockaddr_in*)address)->sin_addr.s_addr;break;caseAF_INET6:addressBytes=((structsockaddr_in6*)address)->sin6_addr.s6_addr;break;default:break;}if(addressBytes&&inet_ntop(family,addressBytes,addressBuffer,BXArraySize(addressBuffer))){NSString*humanReadableAddress=[NSStringstringWithUTF8String:addressBuffer];BXLogInfo(@"Trying '%@'",humanReadableAddress);[connectionDictionarysetObject:humanReadableAddressforKey:kPGTSHostAddressKey];char*conninfo=PGTSCopyConnectionString(connectionDictionary);if([selfstartNegotiation:conninfo]){reachedServer=YES;free(conninfo);break;}free(conninfo);}}}if(reachedServer)[selfnegotiateConnection];else[selffinishedConnecting:NO];}-(void)hostResolverDidFail:(BXHostResolver*)resolvererror:(NSError*)error{[selfsetConnectionError:error];[selffinishedConnecting:NO];}-(void)socketReady:(CFSocketCallBackType)callBackType{BXLogDebug(@"Socket got ready.");//Sometimes the wrong callback type gets called. We cope with this//by checking against an expected type and re-enabling it if needed.if(callBackType!=mExpectedCallBack)CFSocketEnableCallBacks(mSocket,mExpectedCallBack);else{PostgresPollingStatusTypestatus=mPollFunction(mConnection);[selfsetUpSSL];switch(status){casePGRES_POLLING_OK:[selffinishedConnecting:YES];break;casePGRES_POLLING_FAILED:[selffinishedConnecting:NO];break;casePGRES_POLLING_ACTIVE:[selfsocketReady:mExpectedCallBack];break;casePGRES_POLLING_READING:CFSocketEnableCallBacks(mSocket,kCFSocketReadCallBack);mExpectedCallBack=kCFSocketReadCallBack;break;casePGRES_POLLING_WRITING:default:CFSocketEnableCallBacks(mSocket,kCFSocketWriteCallBack);mExpectedCallBack=kCFSocketWriteCallBack;break;}}}-(void)finishedConnecting:(BOOL)succeeded{[selffreeCFTypes];[superfinishedConnecting:succeeded];}#pragma mark Connection methods-(BOOL)connect:(NSDictionary*)connectionDictionary{BXLogDebug(@"Beginning connecting.");BOOLretval=NO;mExpectedCallBack=0;[selfprepareForConnect];[selfsetConnectionDictionary:connectionDictionary];//CFSocket etc. do some nice things for us that prevent libpq from noticing//connection problems. This causes SIGPIPE to be sent to us, and we get//"Broken pipe" as the error message. To cope with this, we check the socket's//status after connecting but before giving it to CFSocket.//For this to work, we need to resolve the host name by ourselves, if we have one.//If the name begins with a slash, it is a path to socket.NSString*name=[connectionDictionaryobjectForKey:kPGTSHostKey];if(0<[namelength]&&'/'!=[namecharacterAtIndex:0]){[mHostResolversetRunLoop:mRunLoop];[mHostResolversetRunLoopMode:(id)kCFRunLoopCommonModes];[mHostResolverresolveHost:name];}else{char*conninfo=PGTSCopyConnectionString(mConnectionDictionary);if([selfstartNegotiation:conninfo]){retval=YES;[selfnegotiateConnection];}else{[selffinishedConnecting:NO];}free(conninfo);}returnretval;}-(BOOL)startNegotiation:(constchar*)conninfo{BXLogDebug(@"Beginning negotiation.");mNegotiationStarted=NO;BOOLretval=NO;if([selfstart:conninfo]){if(CONNECTION_BAD!=PQstatus(mConnection)){mNegotiationStarted=YES;intsocket=PQsocket(mConnection);if(socket<0)BXLogInfo(@"Unable to get connection socket from libpq.");else{//We mimic libpq's error message because it doesn't provide us with error codes.//Also we need to select first to make sure that getsockopt returns an accurate error message.BOOLhaveError=NO;NSString*reason=nil;{charmessage[256]={};intstatus=0;structtimevaltimeout={.tv_sec=15,.tv_usec=0};fd_setmask={};FD_ZERO(&mask);FD_SET(socket,&mask);status=select(socket+1,NULL,&mask,NULL,&timeout);if(status<=0){haveError=YES;strerror_r(errno,message,BXArraySize(message));reason=[NSStringstringWithUTF8String:message];}else{intoptval=0;socklen_tsize=sizeof(optval);status=getsockopt(socket,SOL_SOCKET,SO_ERROR,&optval,&size);if(0==status){if(0==optval)retval=YES;else{haveError=YES;strerror_r(optval,message,BXArraySize(message));reason=[NSStringstringWithUTF8String:message];}}else{haveError=YES;}}}if(haveError){NSString*errorTitle=NSLocalizedStringWithDefaultValue(@"connectionError",nil,[NSBundlebundleForClass:[selfclass]],@"Connection error",@"Title for a sheet.");NSString*messageFormat=NSLocalizedStringWithDefaultValue(@"libpqStyleConnectionErrorFormat",nil,[NSBundlebundleForClass:[selfclass]],@"Could not connect to server: %@. Is the server running at \"%@\" and accepting TCP/IP connections on port %s?",@"Reason for error");if(!reason){reason=NSLocalizedStringWithDefaultValue(@"connectionRefused",nil,[NSBundlebundleForClass:[selfclass]],@"Connection refused",@"Reason for error");}NSString*address=([mConnectionDictionaryobjectForKey:kPGTSHostKey]?:[mConnectionDictionaryobjectForKey:kPGTSHostAddressKey]);NSString*message=[NSStringstringWithFormat:messageFormat,reason,address,PQport(mConnection)];NSDictionary*userInfo=[NSDictionarydictionaryWithObjectsAndKeys:errorTitle,NSLocalizedDescriptionKey,errorTitle,NSLocalizedFailureReasonErrorKey,message,NSLocalizedRecoverySuggestionErrorKey,nil];NSError*error=[BXErrorerrorWithDomain:kPGTSConnectionErrorDomaincode:kPGTSConnectionErrorUnknownuserInfo:userInfo];[selfsetConnectionError:error];}}}}returnretval;}-(void)negotiateConnection{BXLogDebug(@"Negotiating.");if(mTraceFile)PQtrace(mConnection,mTraceFile);CFSocketContextcontext={0,self,NULL,NULL,NULL};CFSocketCallBackTypecallbacks=kCFSocketReadCallBack|kCFSocketWriteCallBack;mSocket=CFSocketCreateWithNative(NULL,PQsocket(mConnection),callbacks,&SocketReady,&context);CFOptionFlagsflags=~kCFSocketAutomaticallyReenableReadCallBack&~kCFSocketAutomaticallyReenableWriteCallBack&~kCFSocketCloseOnInvalidate&CFSocketGetSocketFlags(mSocket);CFSocketSetSocketFlags(mSocket,flags);mSocketSource=CFSocketCreateRunLoopSource(NULL,mSocket,0);BXAssertLog(mSocket,@"Expected source to have been created.");BXAssertLog(CFSocketIsValid(mSocket),@"Expected socket to be valid.");BXAssertLog(mSocketSource,@"Expected socketSource to have been created.");BXAssertLog(CFRunLoopSourceIsValid(mSocketSource),@"Expected socketSource to be valid.");CFSocketDisableCallBacks(mSocket,kCFSocketReadCallBack);CFSocketEnableCallBacks(mSocket,kCFSocketWriteCallBack);mExpectedCallBack=kCFSocketWriteCallBack;CFRunLoopAddSource(mRunLoop,mSocketSource,(CFStringRef)kCFRunLoopCommonModes);}@end@implementationPGTSAsynchronousReconnector-(id)init{if((self=[superinit])){mPollFunction=&PQresetPoll;}returnself;}-(BOOL)start:(constchar*)connectionString{return(BOOL)PQresetStart(mConnection);}@end