text: /*
   *  Phusion Passenger - https://www.phusionpassenger.com/
   *  Copyright (c) 2010-2014 Phusion
   *
   *  "Phusion Passenger" is a trademark of Hongli Lai & Ninh Bui.
   *
   *  Permission is hereby granted, free of charge, to any person obtaining a copy
   *  of this software and associated documentation files (the "Software"), to deal
   *  in the Software without restriction, including without limitation the rights
   *  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   *  copies of the Software, and to permit persons to whom the Software is
   *  furnished to do so, subject to the following conditions:
   *
   *  The above copyright notice and this permission notice shall be included in
   *  all copies or substantial portions of the Software.
   *
   *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
   *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
   *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
   *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
   *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
   *  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
   *  THE SOFTWARE.
   */
  #ifndef _PASSENGER_UNION_STATION_CORE_H_
  #define _PASSENGER_UNION_STATION_CORE_H_
  
  #include <boost/shared_ptr.hpp>
  #include <boost/enable_shared_from_this.hpp>
  #include <boost/thread.hpp>
  #include <oxt/backtrace.hpp>
  
  #include <errno.h>
  
  #include <string>
  #include <vector>
  #include <stdexcept>
  
  #include <Logging.h>
  #include <Exceptions.h>
  #include <RandomGenerator.h>
  #include <StaticString.h>
  #include <UnionStation/Connection.h>
  #include <UnionStation/Transaction.h>
  #include <Utils.h>
  #include <Utils/MessageIO.h>
  #include <Utils/SystemTime.h>
  
  namespace Passenger {
  namespace UnionStation {
  
  using namespace std;
  using namespace boost;
  
  
  class Core: public boost::enable_shared_from_this<Core> {
  private:
  	static const unsigned int CONNECTION_POOL_MAX_SIZE = 10;
  	static const unsigned int TXN_ID_MAX_SIZE =
  		2 * sizeof(unsigned int) +    // max hex timestamp size
  		11 +                          // space for a random identifier
  		1;                            // null terminator
  
  	/**** Server information ****/
  	const string serverAddress;
  	const string username;
  	const string password;
  	const string nodeName;
  
  	/**** Working objects ****/
  	RandomGenerator randomGenerator;
  	TransactionPtr nullTransaction;
  
  	/********************** Connection handling fields **********************
  	 * These fields are synchronized through the mutex. The contents
  	 * of the conntection objects are not synchronized through this mutex,
  	 * but through the Connection object's own mutex.
  	 ************************************************************************/
  	mutable boost::mutex syncher;
  	vector<ConnectionPtr> connectionPool;
  	/** How long to wait before reconnecting. */
  	unsigned long long reconnectTimeout;
  	/** Earliest time at which we should attempt a reconnect. Earlier attempts
  	 * will fail. Calculated from reconnectTimeout.
  	 */
  	unsigned long long nextReconnectTime;
  
  	static string determineNodeName(const string &givenNodeName) {
  		if (givenNodeName.empty()) {
  			return getHostName();
  		} else {
  			return givenNodeName;
  		}
  	}
  
  	static bool isNetworkError(int code) {
  		return code == EPIPE || code == ECONNREFUSED || code == ECONNRESET
  			|| code == EHOSTUNREACH || code == ENETDOWN || code == ENETUNREACH
  			|| code == ETIMEDOUT;
  	}
  
  	template<typename T>
  	static bool instanceof(const std::exception &e) {
  		return dynamic_cast<const T *>(&e) != NULL;
  	}
  
  	void initialize() {
  		nullTransaction   = boost::make_shared<Transaction>();
  		reconnectTimeout  = 1000000;
  		nextReconnectTime = 0;
  	}
  
  	/** Creates a transaction ID string. `txnId` MUST be at least TXN_ID_MAX_SIZE bytes. */
  	void createTxnId(char *txnId, char **txnIdEnd, unsigned long long timestamp) {
  		unsigned int timestampSize;
  		char *end;
  
  		// "[timestamp]"
  		// Our timestamp is like a Unix timestamp but with minutes
  		// resolution instead of seconds. 32 bits will last us for
  		// about 8000 years.
  		timestampSize = integerToHexatri<unsigned int>(
  			timestamp / 1000000 / 60,
  			txnId);
  		end = txnId + timestampSize;
  
  		// "[timestamp]-"
  		*end = '-';
  		end++;
  
  		// "[timestamp]-[random id]"
  		randomGenerator.generateAsciiString(end, 11);
  		end += 11;
  		*end = '\0';
  		*txnIdEnd = end;
  	}
  
  	ConnectionPtr createNewConnection() {
  		TRACE_POINT();
  		int fd;
  		vector<string> args;
  		unsigned long long timeout = 15000000;
  
  		// Create socket.
  		fd = connectToServer(serverAddress);
  		FdGuard guard(fd, true);
  		
  		// Handshake: process protocol version number.
  		if (!readArrayMessage(fd, args, &timeout)) {
  			throw IOException("The logging agent closed the connection before sending a version identifier.");
  		}
  		if (args.size() != 2 || args[0] != "version") {
  			throw IOException("The logging agent server didn't sent a valid version identifier.");
  		}
  		if (args[1] != "1") {
  			string message = "Unsupported logging agent protocol version " +
  				args[1] + ".";
  			throw IOException(message);
  		}
  
  		// Handshake: authenticate.
  		UPDATE_TRACE_POINT();
  		writeScalarMessage(fd, username, &timeout);
  		writeScalarMessage(fd, password, &timeout);
  
  		UPDATE_TRACE_POINT();
  		if (!readArrayMessage(fd, args, &timeout)) {
  			throw IOException("The logging agent did not send an authentication response.");
  		} else if (args.size() != 1) {
  			throw IOException("The authentication response that the logging agent sent is not valid.");
  		} else if (args[0] != "ok") {
  			throw SecurityException("The logging agent server denied authentication: " + args[0]);
  		}
  
  		// Initialize session.
  		UPDATE_TRACE_POINT();
  		writeArrayMessage(fd, &timeout, "init", nodeName.c_str(), NULL);
  		if (!readArrayMessage(fd, args, &timeout)) {
  			throw SystemException("Cannot connect to logging server", ECONNREFUSED);
  		} else if (args.size() != 1) {
  			throw IOException("Logging server returned an invalid reply for the 'init' command");
  		} else if (args[0] == "server shutting down") {
  			throw SystemException("Cannot connect to server", ECONNREFUSED);
  		} else if (args[0] != "ok") {
  			throw IOException("Logging server returned an invalid reply for the 'init' command");
  		}
  
  		ConnectionPtr connection = boost::make_shared<Connection>(fd);
  		guard.clear();
  		return connection;
  	}
  
  public:
  	Core() {
  		initialize();
  	}
  
  	Core(const string &_serverAddress, const string &_username,
  	     const string &_password, const string &_nodeName = string())
  		: serverAddress(_serverAddress),
  		  username(_username),
  		  password(_password),
  		  nodeName(determineNodeName(_nodeName))
  	{
  		initialize();
  	}
  
  
  	/***** Connection pool methods *****/
  
  	ConnectionPtr checkoutConnection() {
  		TRACE_POINT();
  		boost::unique_lock<boost::mutex> l(syncher);
  		if (!connectionPool.empty()) {
  			P_TRACE(3, "Checked out existing connection");
  			ConnectionPtr connection = connectionPool.back();
  			connectionPool.pop_back();
  			return connection;
  
  		} else {
  			if (SystemTime::getUsec() < nextReconnectTime) {
  				P_TRACE(3, "Not yet time to reconnect; returning NULL connection");
  				return ConnectionPtr();
  			}
  
  			l.unlock();
  			P_TRACE(3, "Creating new connection with logging agent");
  			ConnectionPtr connection;
  			try {
  				connection = createNewConnection();
  			} catch (const TimeoutException &) {
  				l.lock();
  				P_WARN("Timeout trying to connect to the logging agent at " << serverAddress << "; " <<
  					"will reconnect in " << reconnectTimeout / 1000000 << " second(s).");
  				nextReconnectTime = SystemTime::getUsec() + reconnectTimeout;
  				return ConnectionPtr();
  			} catch (const tracable_exception &e) {
  				l.lock();
  				nextReconnectTime = SystemTime::getUsec() + reconnectTimeout;
  				if (instanceof<IOException>(e) || instanceof<SystemException>(e)) {
  					P_WARN("Cannot connect to the logging agent at " << serverAddress <<
  						" (" << e.what() << "); will reconnect in " <<
  						reconnectTimeout / 1000000 << " second(s).");
  					return ConnectionPtr();
  				} else {
  					throw;
  				}
  			}
  
  			return connection;
  		}
  	}
  
  	void checkinConnection(const ConnectionPtr &connection) {
  		boost::unique_lock<boost::mutex> l(syncher);
  		if (connectionPool.size() < CONNECTION_POOL_MAX_SIZE) {
  			connectionPool.push_back(connection);
  		} else {
  			l.unlock();
  			connection->disconnect();
  		}
  	}
  
  
  	/***** Transaction methods *****/
  
  	TransactionPtr createNullTransaction() const {
  		return nullTransaction;
  	}
  
  	bool sendRequest(const ConnectionPtr &connection, StaticString args[],
  		unsigned int nargs, bool expectAck)
  	{
  		ConnectionLock cl(connection);
  		ConnectionGuard guard(connection.get());
  
  		try {
  			unsigned long long timeout = 15000000;
  
  			writeArrayMessage(connection->fd, args, nargs, &timeout);
  
  			if (expectAck) {
  				vector<string> args;
  				if (!readArrayMessage(connection->fd, args, &timeout)) {
  					boost::lock_guard<boost::mutex> l(syncher);
  					P_WARN("The logging agent at " << serverAddress <<
  						" closed the connection (no error message given);" <<
  						" will reconnect in " << reconnectTimeout / 1000000 <<
  						" second(s).");
  					nextReconnectTime = SystemTime::getUsec() + reconnectTimeout;
  					return false;
  				} else if (args.size() == 2 && args[0] == "error") {
  					boost::lock_guard<boost::mutex> l(syncher);
  					P_WARN("The logging agent at " << serverAddress <<
  						" closed the connection (error message: " << args[1] <<
  						"); will reconnect in " << reconnectTimeout / 1000000 <<
  						" second(s).");
  					nextReconnectTime = SystemTime::getUsec() + reconnectTimeout;
  					return false;
  				} else if (args.empty() || args[0] != "ok") {
  					boost::lock_guard<boost::mutex> l(syncher);
  					P_WARN("The logging agent at " << serverAddress <<
  						" sent an unexpected reply;" <<
  						" will reconnect in " << reconnectTimeout / 1000000 <<
  						" second(s).");
  					nextReconnectTime = SystemTime::getUsec() + reconnectTimeout;
  					return false;
  				}
  			}
  
  			guard.clear();
  			return true;
  
  		} catch (const TimeoutException &) {
  			boost::lock_guard<boost::mutex> l(syncher);
  			P_WARN("Timeout trying to communicate with the logging agent at " << serverAddress << "; " <<
  				"will reconnect in " << reconnectTimeout / 1000000 << " second(s).");
  			nextReconnectTime = SystemTime::getUsec() + reconnectTimeout;
  			return false;
  
  		} catch (const SystemException &e) {
  			if (e.code() == ENOENT || isNetworkError(e.code())) {
  				string errorResponse;
  				bool gotErrorResponse;
  
  				guard.clear();
  				gotErrorResponse = connection->disconnect(errorResponse);
  				boost::lock_guard<boost::mutex> l(syncher);
  				if (gotErrorResponse) {
  					P_WARN("The logging agent at " << serverAddress <<
  						" closed the connection (error message: " << errorResponse <<
  						"); will reconnect in " << reconnectTimeout / 1000000 <<
  						" second(s).");
  				} else {
  					P_WARN("The logging agent at " << serverAddress <<
  						" closed the connection (no error message given);" <<
  						" will reconnect in " << reconnectTimeout / 1000000 <<
  						" second(s).");
  				}
  				nextReconnectTime = SystemTime::getUsec() + reconnectTimeout;
  				return false;
  			} else {
  				throw;
  			}
  		}
  	}
  
  	TransactionPtr newTransaction(const string &groupName,
  		const string &category = "requests",
  		const string &unionStationKey = "-",
  		const string &filters = string())
  	{
  		if (isNull()) {
  			return createNullTransaction();
  		}
  
  		// Prepare parameters.
  		unsigned long long timestamp = SystemTime::getUsec();
  		char txnId[TXN_ID_MAX_SIZE], *txnIdEnd;
  		char timestampStr[2 * sizeof(unsigned long long) + 1];
  
  		createTxnId(txnId, &txnIdEnd, timestamp);
  		integerToHexatri<unsigned long long>(timestamp, timestampStr);
  
  		StaticString params[] = {
  			StaticString("openTransaction", sizeof("openTransaction") - 1),
  			StaticString(txnId, txnIdEnd - txnId),
  			groupName,
  			// empty nodeName, implies using the default
  			// nodeName passed during initialization
  			StaticString(),
  			category,
  			timestampStr,
  			unionStationKey,
  			StaticString("true", 4), // crashProtect
  			StaticString("true", 4), // ack
  			filters
  		};
  		unsigned int nparams = sizeof(params) / sizeof(StaticString);
  
  		// Get a connection to the logging server.
  		ConnectionPtr connection = checkoutConnection();
  		if (connection == NULL) {
  			P_TRACE(2, "Created NULL Union Station transaction: group=" << groupName <<
  				", category=" << category << ", txnId=" <<
  				StaticString(txnId, txnIdEnd - txnId));
  			return createNullTransaction();
  		}
  
  		// Send request, process reply.
  		if (sendRequest(connection, params, nparams, true)) {
  			ConnectionGuard guard(connection.get());
  			TransactionPtr transaction = boost::make_shared<Transaction>(
  				shared_from_this(),
  				connection,
  				string(txnId, txnIdEnd - txnId),
  				groupName,
  				category,
  				unionStationKey);
  			guard.clear();
  			P_TRACE(2, "Created new Union Station transaction: group=" << groupName <<
  				", category=" << category << ", txnId=" <<
  				StaticString(txnId, txnIdEnd - txnId));
  			return transaction;
  		} else {
  			P_TRACE(2, "Created NULL Union Station transaction: group=" << groupName <<
  				", category=" << category << ", txnId=" <<
  				StaticString(txnId, txnIdEnd - txnId));
  			return createNullTransaction();
  		}
  	}
  
  	TransactionPtr continueTransaction(const string &txnId,
  		const string &groupName,
  		const string &category = "requests",
  		const string &unionStationKey = "-")
  	{
  		if (isNull() || txnId.empty()) {
  			return createNullTransaction();
  		}
  
  		// Prepare parameters.
  		char timestampStr[2 * sizeof(unsigned long long) + 1];
  		integerToHexatri<unsigned long long>(SystemTime::getUsec(), timestampStr);
  
  		StaticString params[] = {
  			StaticString("openTransaction", sizeof("openTransaction") - 1),
  			txnId,
  			groupName,
  			// empty nodeName, implies using the default
  			// nodeName passed during initialization
  			StaticString(),
  			category,
  			timestampStr,
  			unionStationKey,
  			StaticString("true", 4),  // crashProtect
  			StaticString("false", 4)  // ack
  		};
  		unsigned int nparams = sizeof(params) / sizeof(StaticString);
  
  		// Get a connection to the logging server.
  		ConnectionPtr connection = checkoutConnection();
  		if (connection == NULL) {
  			return createNullTransaction();
  		}
  
  		// Send request.
  		if (sendRequest(connection, params, nparams, false)) {
  			ConnectionGuard guard(connection.get());
  			TransactionPtr transaction = boost::make_shared<Transaction>(
  				shared_from_this(),
  				connection,
  				txnId,
  				groupName,
  				category,
  				unionStationKey);
  			guard.clear();
  			return transaction;
  		} else {
  			return createNullTransaction();
  		}
  	}
  
  
  	/***** Parameter getters and setters *****/
  
  	void setReconnectTimeout(unsigned long long usec) {
  		boost::lock_guard<boost::mutex> l(syncher);
  		reconnectTimeout = usec;
  	}
  
  	bool isNull() const {
  		return serverAddress.empty();
  	}
  
  	const string &getAddress() const {
  		return serverAddress;
  	}
  
  	const string &getUsername() const {
  		return username;
  	}
  
  	const string &getPassword() const {
  		return password;
  	}
  
  	/**
  	 * @post !result.empty()
  	 */
  	const string &getNodeName() const {
  		return nodeName;
  	}
  };
  
  
  inline void
  _checkinConnection(const CorePtr &core, const ConnectionPtr &connection) {
  	core->checkinConnection(connection);
  }
  
  
  } // namespace UnionStation
  } // namespace Passenger
  
  #endif /* _PASSENGER_UNION_STATION_CORE_H_ */
  