Class for maintaining a Pool of Open FTP sessions
| 253 | local_file.close() |
| 254 | |
| 255 | class FTP_pool: |
| 256 | """ |
| 257 | Class for maintaining a Pool of Open FTP sessions |
| 258 | """ |
| 259 | def __init__(self, module_id, config): |
| 260 | self.number_of_connections = int(config.find_config("NUM_CONNECTION")) |
| 261 | self.wait_for_connnection_time = int(config.find_config("FTP_TIMEOUT")) |
| 262 | self.ftp_session_pool = [] |
| 263 | self.ftp_pool_lock = [] |
| 264 | self.module_id = module_id |
| 265 | self.machine_name = config.find_ftp_info_machine_name_for_module(module_id) |
| 266 | self.uid = config.find_ftp_info_for_module(module_id, 'ftp_uid') |
| 267 | self.pwd = config.find_ftp_info_for_module(module_id, 'ftp_password') |
| 268 | self.thread_list = [] |
| 269 | for i in range(self.number_of_connections): |
| 270 | logger.debug("Creating FTP session " + str(i) +" "+ self.machine_name+" "+ self.uid+" "+self.pwd) |
| 271 | ftp_session = ftplib.FTP(self.machine_name) |
| 272 | ftp_session.login(self.uid, self.pwd) |
| 273 | self.ftp_session_pool.append(ftp_session) |
| 274 | self.ftp_pool_lock.append(threading.Lock()) |
| 275 | |
| 276 | def __get_ftp_session_id_(self): |
| 277 | """ Returns an unsued FTP session from the pool |
| 278 | """ |
| 279 | wait_time = 0 |
| 280 | while True: |
| 281 | for i in range(self.number_of_connections): |
| 282 | if self.ftp_pool_lock[i].locked() == False: |
| 283 | self.ftp_pool_lock[i].acquire() |
| 284 | logger.debug("Request for ftp session got session #" + str(i) ) |
| 285 | return i |
| 286 | time.sleep(3) |
| 287 | wait_time = wait_time + 3 |
| 288 | if wait_time < self.wait_for_connnection_time: |
| 289 | logger.debug("No ftp sessions available..re-checking after wait..." ) |
| 290 | continue |
| 291 | else: |
| 292 | raise NameError, "TimeOut for FTP" |
| 293 | |
| 294 | def release_ftp_session_id(self, ftp_session_id): |
| 295 | """ Method to release the FTP session back to the common pool |
| 296 | """ |
| 297 | self.ftp_pool_lock[ftp_session_id].release() |
| 298 | |
| 299 | |
| 300 | def asynchronous_ftp_get(self,remote_file,localfile): |
| 301 | """ This function is called to invoke a thread that will download the file |
| 302 | """ |
| 303 | ftp_session_id = self.__get_ftp_session_id_() |
| 304 | logger.debug("session allocated " + str(ftp_session_id)) |
| 305 | thread_id = ftp_thread( self.ftp_session_pool[ftp_session_id],ftp_session_id,remote_file, localfile, self) |
| 306 | thread_id.start() |
| 307 | self.thread_list.append(thread_id) |
| 308 | |
| 309 | def close_all(self): |
| 310 | """ Close all FTP connections and threads |
| 311 | """ |
| 312 | for tid in self.thread_list: |