# $SnapHashLicense:
#
# SnapLogic - Open source data services
#
# Copyright (C) 2009, SnapLogic, Inc. All rights reserved.
#
# See http://www.snaplogic.org for more information about
# the SnapLogic project.
#
# This program is free software, distributed under the terms of
# the GNU General Public License Version 2. See the LEGAL file
# at the top of the source tree.
#
# "SnapLogic" is a trademark of SnapLogic, Inc.
#
#
# $
# $Id: stream_driver.py 9150 2009-10-06 16:34:22Z dmitri $
"""
Contains HttpPipeDriver used to drive HTTP connections for use with SelectablePipe.
"""
from threading import Thread
import select
from snaplogic.common.snap_exceptions import *
class StreamDriver(Thread):
"""
Read data from a URL and send it to the input view of a resource in local pipeline.
"""
def __init__(self, stream, logger = None):
"""
Initialize the thread for the stream driver.
Takes the input stream which has the connection and the pipe stored in it.
Connection is typically the HttpResponse (source from which this driver
will be reading the data), and the pipe is the destination pipe
(destination where data is sent -- resource input view in the local pipeline).
"""
super(StreamDriver, self).__init__()
self.name = "StreamDriver"
self._logger = logger
self._stream = stream
def run(self):
"""
The body of the stream driver thread.
Read data from input stream and send it down the selectable pipe.
Typically we will be reading data from a URL, e.g. remote URL from another CC,
and sending it via the selectable pipe to one of the local resources in a pipeline.
Once we finish reading close the pipe.
"""
while True:
try:
data = self._stream._conn.read()
if not data:
self._stream._pipe.close()
break
self._stream._pipe.put(data)
except Exception, e:
if self._logger:
self._logger.elog(e)
try:
self._stream._pipe.abort()
except Exception, e:
if self._logger:
self._logger.elog(e)
try:
self._stream._conn.close()
except Exception, e:
if self._logger:
self._logger.elog(e)
def stop(self, timeout=None):
"""
This method is not called by our product code, only by QA tests.
If there is a need this method may be implemented in the future
"""
pass
|