X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=application%2Fhub%2Fmain%2Fhelper%2Fconnection%2Fclass_BaseConnectionHelper.php;h=0b5a0ab871185cb920611b7a7d0627b05874b520;hb=d1cefc7b88d3e7f80ad6f13c548fc0dbfc59bdf1;hp=4771e4d669b0531472f802200c76efaa38760488;hpb=87e41bdad425088c7e85bdef0679283e90694fca;p=hub.git diff --git a/application/hub/main/helper/connection/class_BaseConnectionHelper.php b/application/hub/main/helper/connection/class_BaseConnectionHelper.php index 4771e4d66..0b5a0ab87 100644 --- a/application/hub/main/helper/connection/class_BaseConnectionHelper.php +++ b/application/hub/main/helper/connection/class_BaseConnectionHelper.php @@ -4,7 +4,7 @@ * * @author Roland Haeder * @version 0.0.0 - * @copyright Copyright (c) 2007, 2008 Roland Haeder, 2009 - 2011 Hub Developer Team + * @copyright Copyright (c) 2007, 2008 Roland Haeder, 2009 - 2012 Hub Developer Team * @license GNU GPL 3.0 or any newer version * @link http://www.ship-simu.org * @@ -22,6 +22,24 @@ * along with this program. If not, see . */ class BaseConnectionHelper extends BaseHubHelper implements Registerable, ProtocolHandler { + // Exception codes + const EXCEPTION_UNSUPPORTED_ERROR_HANDLER = 0x900; + + /** + * Connection type 'incoming' + */ + const CONNECTION_TYPE_INCOMING = 'incoming'; + + /** + * Connection type 'outgoing' + */ + const CONNECTION_TYPE_OUTGOING = 'outgoing'; + + /** + * Connection type 'server' + */ + const CONNECTION_TYPE_SERVER = 'server'; + /** * Protocol used */ @@ -43,17 +61,17 @@ class BaseConnectionHelper extends BaseHubHelper implements Registerable, Protoc private $sentData = 0; /** - * Offset + * Difference */ - private $offset = 0; + private $diff = 0; /** - * Connect retries for this connection + * Whether this connection is initialized */ - private $retryCount = 0; + private $isInitialized = false; /** - * Wether this connection is shutted down + * Whether this connection is shutted down */ private $shuttedDown = false; @@ -77,8 +95,36 @@ class BaseConnectionHelper extends BaseHubHelper implements Registerable, Protoc // Call parent constructor parent::__construct($className); + // Initialize output stream + $streamInstance = ObjectFactory::createObjectByConfiguredName('node_raw_data_output_stream_class'); + + // And add it to this connection helper + $this->setOutputStreamInstance($streamInstance); + + // Init state which sets the state to 'init' + $this->initState(); + // Register this connection helper Registry::getRegistry()->addInstance('connection', $this); + + // Get the fragmenter instance + $fragmenterInstance = FragmenterFactory::createFragmenterInstance('package'); + + // Set it here + $this->setFragmenterInstance($fragmenterInstance); + } + + /** + * Getter for real class name, overwrites generic method and is final + * + * @return $class Name of this class + */ + public final function __toString () { + // Class name representation + $class = self::getConnectionClassName($this->getAddress(), $this->getPort(), parent::__toString()); + + // Return it + return $class; } /** @@ -138,6 +184,134 @@ class BaseConnectionHelper extends BaseHubHelper implements Registerable, Protoc $this->address = $address; } + /** + * Initializes the current connection + * + * @return void + * @throws SocketOptionException If setting any socket option fails + */ + protected function initConnection () { + // Get socket resource + $socketResource = $this->getSocketResource(); + + // Set the option to reuse the port + if (!socket_set_option($socketResource, SOL_SOCKET, SO_REUSEADDR, 1)) { + // Handle this socket error with a faked recipientData array + $this->handleSocketError(__METHOD__, __LINE__, $socketResource, array('0.0.0.0', '0')); + + // And throw again + // @TODO Move this to the socket error handler + throw new SocketOptionException(array($this, $socketResource, $socketError, $errorMessage), BaseListener::EXCEPTION_INVALID_SOCKET); + } // END - if + + /* + * Set socket to non-blocking mode before trying to establish a link to + * it. This is now the default behaviour for all connection helpers who + * call initConnection(); . + */ + if (!socket_set_nonblock($socketResource)) { + // Handle this socket error with a faked recipientData array + $helperInstance->handleSocketError(__METHOD__, __LINE__, $socketResource, array('0.0.0.0', '0')); + + // And throw again + throw new SocketOptionException(array($helperInstance, $socketResource, $socketError, $errorMessage), BaseListener::EXCEPTION_INVALID_SOCKET); + } // END - if + + // Last step: mark connection as initialized + $this->isInitialized = true; + } + + /** + * Attempts to connect to a peer by given IP number and port from a valid + * recipientData array with currently configured timeout. + * + * @param $recipientData A valid recipient data array, 0=IP; 1=PORT + * @return $isConnected Whether the connection went fine + * @see Please see http://de.php.net/manual/en/function.socket-connect.php#84465 for original code + * @todo Rewrite the while() loop to a iterator to not let the software stay very long here + */ + protected function connectToPeerByRecipientData (array $recipientData) { + // Only call this if the connection is initialized by initConnection() + assert($this->isInitialized === true); + + // Get current time + $time = time(); + + // "Cache" socket resource and timeout config + $socketResource = $this->getSocketResource(); + $timeout = $this->getConfigInstance()->getConfigEntry('socket_timeout_seconds'); + + // Debug output + self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: Trying to connect to ' . $recipientData[0] . ':' . $recipientData[1] . ' with socketResource[' . gettype($socketResource) . ']=' . $socketResource . ' ...'); + + // Try to connect until it is connected + while ($isConnected = !@socket_connect($socketResource, $recipientData[0], $recipientData[1])) { + // Get last socket error + $socketError = socket_last_error($socketResource); + + // Skip any errors which may happen on non-blocking connections + if (($socketError == SOCKET_EINPROGRESS) || ($socketError == SOCKET_EALREADY)) { + // Now, is that attempt within parameters? + if ((time() - $time) >= $timeout) { + // Didn't work within timeout + $isConnected = false; + break; + } // END - if + + // Sleep about one second + $this->idle(1000); + } elseif ($socketError != 0) { + // Stop on everything else pronto + $isConnected = false; + break; + } + } // END - while + + // Is the peer connected? + if ($isConnected === true) { + // Connection is fully established here, so change the state. + PeerStateFactory::createPeerStateInstanceByName('connected', $this); + } else { + /* + * There was a problem connecting to the peer (this state is a meta + * state until the error handler has found the real cause). + */ + PeerStateFactory::createPeerStateInstanceByName('problem', $this); + } + + // Return status + return $isConnected; + } + + /** + * Static "getter" for this connection class' name + * + * @param $address IP address + * @param $port Port number + * @param $className Original class name + * @return $class Expanded class name + */ + public static function getConnectionClassName ($address, $port, $className) { + // Construct it + $class = $address . ':' . $port . ':' . $className; + + // ... and return it + return $class; + } + + /** + * Initializes the peer's state which sets it to 'init' + * + * @return void + */ + private function initState() { + /* + * Get the state factory and create the initial state, we don't need + * the state instance here + */ + PeerStateFactory::createPeerStateInstanceByName('init', $this); + } + /** * "Getter" for raw data from a package array. A fragmenter is used which * will returns us only so many raw data which fits into the back buffer. @@ -154,33 +328,37 @@ class BaseConnectionHelper extends BaseHubHelper implements Registerable, Protoc * @return $chunkData Raw data chunk */ private function getRawDataFromPackageArray (array $packageData) { - // If there is no fragmenter? - if (!Registry::getRegistry()->instanceExists('package_fragmenter')) { - // Get the fragmenter instance - $fragmenterInstance = ObjectFactory::createObjectByConfiguredName('package_fragmenter_class'); - - // Add it to the registry - Registry::getRegistry()->addInstance('package_fragmenter', $fragmenterInstance); - } else { - // Get fragmenter from registry - $fragmenterInstance = Registry::getRegistry()->getInstance('package_fragmenter'); - } - // Implode the package data array and fragement the resulting string, returns the final hash - $this->currentFinalHash = $fragmenterInstance->fragmentPackageArray($packageData, $this); + $finalHash = $this->getFragmenterInstance()->fragmentPackageArray($packageData, $this); + if ($finalHash !== true) { + $this->currentFinalHash = $finalHash; + } // END - if + + // Debug message + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: currentFinalHash=' . $this->currentFinalHash); // Get the next raw data chunk from the fragmenter - $rawDataChunk = $fragmenterInstance->getNextRawDataChunk($this->currentFinalHash); + $rawDataChunk = $this->getFragmenterInstance()->getNextRawDataChunk($this->currentFinalHash); // Get chunk hashes and chunk data $chunkHashes = array_keys($rawDataChunk); $chunkData = array_values($rawDataChunk); - // Remember this chunk as queued - $this->queuedChunks[$chunkHashes[0]] = $chunkData[0]; + // Is the required data there? + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: chunkHashes[]=' . count($chunkHashes) . ',chunkData[]=' . count($chunkData)); + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('chunkData='.print_r($chunkData,true)); + if ((isset($chunkHashes[0])) && (isset($chunkData[0]))) { + // Remember this chunk as queued + $this->queuedChunks[$chunkHashes[0]] = $chunkData[0]; - // Return the raw data - return $chunkData[0]; + // Return the raw data + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: Returning ' . strlen($chunkData[0]) . ' bytes from ' . __METHOD__ . ' ...'); + return $chunkData[0]; + } else { + // Return zero string + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: Returning zero bytes from ' . __METHOD__ . '!'); + return ''; + } } /** @@ -197,99 +375,258 @@ class BaseConnectionHelper extends BaseHubHelper implements Registerable, Protoc /** * Sends raw package data to the recipient * - * @param $packageData Raw package data - * @return $sentBytes Actual sent bytes to the peer + * @param $packageData Raw package data + * @return $totalSentBytes Total sent bytes to the peer * @throws InvalidSocketException If we got a problem with this socket */ public function sendRawPackageData (array $packageData) { - // Convert the package data array to a raw data stream - $rawData = $this->getRawDataFromPackageArray($packageData); + // The helper's state must be 'connected' + $this->getStateInstance()->validatePeerStateConnected(); + + // Reset serial number + $this->getFragmenterInstance()->resetSerialNumber(); + + // Cache buffer length + $bufferSize = $this->getConfigInstance()->getConfigEntry($this->getProtocol() . '_buffer_length'); + + // Init variables + $rawData = ''; + $dataStream = ' '; + $totalSentBytes = 0; + + // Fill sending buffer with data + while (strlen($dataStream) > 0) { + // Convert the package data array to a raw data stream + $dataStream = $this->getRawDataFromPackageArray($packageData); + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: Adding ' . strlen($dataStream) . ' bytes to the sending buffer ...'); + $rawData .= $dataStream; + } // END - while + + // Nothing to sent is bad news, so assert on it + assert(strlen($rawData) > 0); + + // Encode the raw data with our output-stream + $encodedData = $this->getOutputStreamInstance()->streamData($rawData); + + // Debug message + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: rawData()=' . strlen($rawData) . ',encodedData()=' . strlen($encodedData)); + + // Calculate difference + $this->diff = $bufferSize - strlen($encodedData); // Get socket resource $socketResource = $this->getSocketResource(); - // And deliver it - $sentBytes = @socket_write($socketResource, $rawData, $this->getConfigInstance()->getConfigEntry($this->getProtocol() . '_buffer_length') - $this->offset); + // Init sent bytes + $sentBytes = 0; + + // Deliver all data + while ($sentBytes !== false) { + // And deliver it + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: Sending out ' . strlen($encodedData) . ' bytes,bufferSize=' . $bufferSize . ',diff=' . $this->diff); + + if ($this->diff >= 0) { + // Send all out (encodedData is smaller than or equal buffer size) + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: MD5=' . md5(substr($encodedData, 0, ($bufferSize - $this->diff)))); + $sentBytes = socket_write($socketResource, $encodedData, ($bufferSize - $this->diff)); + } else { + // Send buffer size out + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: MD5=' . md5(substr($encodedData, 0, $bufferSize))); + $sentBytes = socket_write($socketResource, $encodedData, $bufferSize); + } + + // If there was an error, we don't continue here + if ($sentBytes === false) { + // Handle the error with a faked recipientData array + $this->handleSocketError(__METHOD__, __LINE__, $socketResource, array('0.0.0.0', '0')); + + // And throw it + throw new InvalidSocketException(array($this, $socketResource, $socketError, $errorMessage), BaseListener::EXCEPTION_INVALID_SOCKET); + } elseif (($sentBytes == 0) && (strlen($encodedData) > 0)) { + // Nothing sent means we are done + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: All sent! (LINE=' . __LINE__ . ')'); + break; + } + + // The difference between sent bytes and length of raw data should not go below zero + assert((strlen($encodedData) - $sentBytes) >= 0); + + // Add total sent bytes + $totalSentBytes += $sentBytes; + + // Cut out the last unsent bytes + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: Sent out ' . $sentBytes . ' of ' . strlen($encodedData) . ' bytes ...'); + $encodedData = substr($encodedData, $sentBytes); + + // Calculate difference again + $this->diff = $bufferSize - strlen($encodedData); + + // Can we abort? + if (strlen($encodedData) <= 0) { + // Abort here, all sent! + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: All sent! (LINE=' . __LINE__ . ')'); + break; + } // END - if + } // END - while - // If there was an error, we don't continue here - if ($sentBytes === false) { - // Get socket error code for verification - $socketError = socket_last_error($socketResource); + // Return sent bytes + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: totalSentBytes=' . $totalSentBytes . ',diff=' . $this->diff); + return $totalSentBytes; + } - // Get error message - $errorMessage = socket_strerror($socketError); + /** + * Marks this connection as shutted down + * + * @return void + */ + protected final function markConnectionShuttedDown () { + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: ' . $this->__toString() . ' has been marked as shutted down'); + $this->shuttedDown = true; - // Shutdown this socket - $this->shutdownSocket($socketResource); + // And remove the (now invalid) socket + $this->setSocketResource(false); + } - // And throw it - throw new InvalidSocketException(array($this, gettype($socketResource), $socketError, $errorMessage), BaseListener::EXCEPTION_INVALID_SOCKET); - } elseif ($sentBytes == 0) { - // Nothing sent is bad news - die(__METHOD__.': Unhandled 0 sent bytes! rawData[]=' . strlen($rawData)); - } + /** + * Getter for shuttedDown + * + * @return $shuttedDown Whether this connection is shutted down + */ + public final function isShuttedDown () { + //* NOISY-DEBUG: */ self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: ' . $this->__toString() . ',shuttedDown=' . intval($this->shuttedDown)); + return $this->shuttedDown; + } - // The difference between sent bytes and length of raw data should not be below zero - assert((strlen($rawData) - $sentBytes) >= 0); + // ************************************************************************ + // Socket error handler call-back methods + // ************************************************************************ - // Return sent bytes - return $sentBytes; + /** + * Handles socket error 'connection timed out', but does not clear it for + * later debugging purposes. + * + * @param $socketResource A valid socket resource + * @param $recipientData An array with two elements: 0=IP number, 1=port number + * @return void + * @throws SocketConnectionException The connection attempts fails with a time-out + */ + protected function socketErrorConnectionTimedOutHandler ($socketResource, array $recipientData) { + // Get socket error code for verification + $socketError = socket_last_error($socketResource); + + // Get error message + $errorMessage = socket_strerror($socketError); + + // Shutdown this socket + $this->shutdownSocket($socketResource); + + // Throw it again + throw new SocketConnectionException(array($this, $socketResource, $socketError, $errorMessage), BaseListener::EXCEPTION_INVALID_SOCKET); } /** - * Getter for real class name + * Handles socket error 'resource temporary unavailable', but does not + * clear it for later debugging purposes. * - * @return $class Name of this class + * @param $socketResource A valid socket resource + * @param $recipientData An array with two elements: 0=IP number, 1=port number + * @return void + * @throws SocketConnectionException The connection attempts fails with a time-out */ - public function __toString () { - // Class name representation - $class = $this->getAddress() . ':' . $this->getPort() . ':' . parent::__toString(); + protected function socketErrorResourceUnavailableHandler ($socketResource, array $recipientData) { + // Get socket error code for verification + $socketError = socket_last_error($socketResource); - // Return it - return $class; + // Get error message + $errorMessage = socket_strerror($socketError); + + // Shutdown this socket + $this->shutdownSocket($socketResource); + + // Throw it again + throw new SocketConnectionException(array($this, $socketResource, $socketError, $errorMessage), BaseListener::EXCEPTION_INVALID_SOCKET); } /** - * Checks wether the connect retry is exhausted + * Handles socket error 'connection refused', but does not clear it for + * later debugging purposes. * - * @return $isExhaused Wether connect retry is exchausted + * @param $socketResource A valid socket resource + * @param $recipientData An array with two elements: 0=IP number, 1=port number + * @return void + * @throws SocketConnectionException The connection attempts fails with a time-out */ - public final function isConnectRetryExhausted () { - // Construct config entry - $configEntry = $this->getProtocol() . '_connect_retry_max'; + protected function socketErrorConnectionRefusedHandler ($socketResource, array $recipientData) { + // Get socket error code for verification + $socketError = socket_last_error($socketResource); - // Check it out - $isExhausted = ($this->retryCount >= $this->getConfigInstance()->getConfigEntry($configEntry)); + // Get error message + $errorMessage = socket_strerror($socketError); - // Return it - return $isExhausted; + // Shutdown this socket + $this->shutdownSocket($socketResource); + + // Throw it again + throw new SocketConnectionException(array($this, $socketResource, $socketError, $errorMessage), BaseListener::EXCEPTION_INVALID_SOCKET); } /** - * Increases the connect retry count + * Handles socket error 'no route to host', but does not clear it for later + * debugging purposes. * + * @param $socketResource A valid socket resource + * @param $recipientData An array with two elements: 0=IP number, 1=port number * @return void + * @throws SocketConnectionException The connection attempts fails with a time-out */ - public final function increaseConnectRetry () { - $this->retryCount++; + protected function socketErrorNoRouteToHostHandler ($socketResource, array $recipientData) { + // Get socket error code for verification + $socketError = socket_last_error($socketResource); + + // Get error message + $errorMessage = socket_strerror($socketError); + + // Shutdown this socket + $this->shutdownSocket($socketResource); + + // Throw it again + throw new SocketConnectionException(array($this, $socketResource, $socketError, $errorMessage), BaseListener::EXCEPTION_INVALID_SOCKET); } /** - * Marks this connection as shutted down + * Handles socket error 'operation already in progress' which happens in + * method connectToPeerByRecipientData() on timed out connection + * attempts. * + * @param $socketResource A valid socket resource + * @param $recipientData An array with two elements: 0=IP number, 1=port number * @return void + * @throws SocketConnectionException The connection attempts fails with a time-out */ - protected final function markConnectionShutdown () { - $this->shuttedDown = true; + protected function socketErrorOperationAlreadyProgressHandler ($socketResource, array $recipientData) { + // Get socket error code for verification + $socketError = socket_last_error($socketResource); + + // Get error message + $errorMessage = socket_strerror($socketError); + + // Half-shutdown this socket (see there for difference to shutdownSocket()) + $this->halfShutdownSocket($socketResource); + + // Throw it again + throw new SocketConnectionException(array($this, $socketResource, $socketError, $errorMessage), BaseListener::EXCEPTION_INVALID_SOCKET); } /** - * Getter for shuttedDown + * Handles socket "error" 'operation now in progress' which can be safely + * passed on with non-blocking connections. * - * @return $shuttedDown Wether this connection is shutted down + * @param $socketResource A valid socket resource + * @param $recipientData An array with two elements: 0=IP number, 1=port number + * @return void */ - public final function isShuttedDown () { - return $this->shuttedDown; + protected function socketErrorOperationInProgressHandler ($socketResource, array $recipientData) { + self::createDebugInstance(__CLASS__)->debugOutput('CONNECTION-HELPER: Operation is now in progress, this is usual for non-blocking connections and is no bug.'); } }