| 152 | self._buffer = cStringIO.StringIO() |
| 153 | |
| 154 | class SubprocessSession(object): |
| 155 | def __init__(self, command, writer, reader, start=True, **kwargs): |
| 156 | self._command = command |
| 157 | self._writer = writer |
| 158 | self._reader = reader |
| 159 | self._kwargs = {'stdin': subprocess.PIPE, 'stdout': subprocess.PIPE} |
| 160 | self._kwargs.update(kwargs) |
| 161 | |
| 162 | self._lock = threading.RLock() |
| 163 | if start: |
| 164 | self.start() |
| 165 | |
| 166 | def __del__(self): |
| 167 | self.close() |
| 168 | |
| 169 | def put(self, frame, request=True): |
| 170 | return self._communicator.put(frame, request) |
| 171 | |
| 172 | def start(self): |
| 173 | with self._lock: |
| 174 | if hasattr(self, '_process'): |
| 175 | return |
| 176 | self._process = subprocess.Popen(self._command, **self._kwargs) |
| 177 | self._communicator = Communicator( |
| 178 | self._writer(self._process.stdin), self._writer.SENTINEL, |
| 179 | self._reader(self._process.stdout), self._reader.SENTINEL |
| 180 | ) |
| 181 | |
| 182 | def close(self, timeout=None): |
| 183 | with self._lock: |
| 184 | if not hasattr(self, '_process'): |
| 185 | return |
| 186 | try: |
| 187 | self._communicator.close(timeout) |
| 188 | finally: |
| 189 | if self._process.poll() is None: |
| 190 | self._process.kill() |
| 191 | del self._process |
| 192 | del self._communicator |
| 193 | |
| 194 | # I use the part above as a generic utility module. A simple echo example follows: |
| 195 | |