1 /* Copyright (C) 2014 InfiniDB, Inc. 2 3 This program is free software; you can redistribute it and/or 4 modify it under the terms of the GNU General Public License 5 as published by the Free Software Foundation; version 2 of 6 the License. 7 8 This program is distributed in the hope that it will be useful, 9 but WITHOUT ANY WARRANTY; without even the implied warranty of 10 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 11 GNU General Public License for more details. 12 13 You should have received a copy of the GNU General Public License 14 along with this program; if not, write to the Free Software 15 Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, 16 MA 02110-1301, USA. */ 17 18 /********************************************************************* 19 * $Id: clientrotator.h 9210 2013-01-21 14:10:42Z rdempsey $ 20 * 21 * 22 ***********************************************************************/ 23 24 /** @file */ 25 26 #ifndef CLIENTROTATOR_H 27 #define CLIENTROTATOR_H 28 29 #include <iostream> 30 #include <vector> 31 #include <boost/thread.hpp> 32 #include <stdint.h> 33 #include <string> 34 35 #include "bytestream.h" 36 #include "messagequeue.h" 37 #include "configcpp.h" 38 39 namespace execplan 40 { 41 42 /** @brief connection handle structure */ 43 class ClientRotator 44 { 45 public: 46 /** @brief ctor 47 */ 48 ClientRotator(uint32_t sid, const std::string& name, bool localQuery = false); 49 50 /** @brief dtor 51 */ ~ClientRotator()52 ~ClientRotator() 53 { 54 if (fClient) 55 { 56 fClient->shutdown(); 57 delete fClient; 58 } 59 } 60 61 /** @brief connnect 62 * 63 * Try connecting to client based on session id. If no connection, 64 * try connectList. 65 * @param timeout in seconds. 66 */ 67 void connect(double timeout = 50); 68 69 /** @brief write 70 * 71 * Write msg to fClient. If unsuccessful, get new connection with 72 * connectList and write. 73 */ 74 void write(const messageqcpp::ByteStream& msg); 75 76 /** @brief shutdown 77 */ shutdown()78 void shutdown() 79 { 80 if (fClient) 81 { 82 fClient->shutdown(); 83 delete fClient; 84 fClient = 0; 85 } 86 } 87 88 /** @brief read 89 */ 90 messageqcpp::ByteStream read(); 91 92 /** @brief getClient 93 */ getClient()94 messageqcpp::MessageQueueClient* getClient() const 95 { 96 return fClient; 97 } 98 99 /** @brief getSessionId 100 */ getSessionId()101 uint32_t getSessionId() const 102 { 103 return fSessionId; 104 } 105 106 /** @brief setSessionId 107 */ setSessionId(uint32_t sid)108 void setSessionId(uint32_t sid) 109 { 110 fSessionId = sid; 111 } 112 113 friend std::ostream& operator<<(std::ostream& output, const ClientRotator& rhs); 114 115 /** @brief reset fClient */ 116 void resetClient(); 117 localQuery()118 bool localQuery() 119 { 120 return fLocalQuery; 121 } localQuery(bool localQuery)122 void localQuery(bool localQuery) 123 { 124 fLocalQuery = localQuery; 125 } 126 static std::string getModule(); 127 128 private: 129 130 //Not copyable 131 ClientRotator(const ClientRotator& ); 132 ClientRotator& operator=(const ClientRotator& ); 133 134 /** @brief load Clients 135 * 136 * Put all entries for client name tag from config file into client list 137 */ 138 void loadClients(); 139 140 /** @brief execute connect 141 * 142 * Make connection and return success. 143 */ 144 bool exeConnect(const std::string& clientName ); 145 146 /** @brief connnect to list 147 * 148 * Try connecting to next client on list 149 * until timeout lapses. Then throw exception. 150 */ 151 void connectList(double timeout = 0.005); 152 153 /** @brief write to message log 154 * 155 * writes message with file name to debug or 156 * critical log. 157 */ 158 void writeToLog(int line, const std::string& msg, bool critical) const; 159 160 const std::string fName; 161 uint32_t fSessionId; 162 messageqcpp::MessageQueueClient* fClient; 163 typedef std::vector<std::string> ClientList; 164 ClientList fClients; 165 config::Config* fCf; 166 int fDebug; 167 boost::mutex fClientLock; 168 bool fLocalQuery; 169 }; 170 171 172 } // namespace 173 #endif 174 // vim:ts=4 sw=4: 175 176