__all__ = 'create_subprocess_exec', 'create_subprocess_shell'

import subprocess
import warnings

from . import events
from . import protocols
from . import streams
from . import tasks
from .log import logger


PIPE = subprocess.PIPE
STDOUT = subprocess.STDOUT
DEVNULL = subprocess.DEVNULL


class SubprocessStreamProtocol(streams.FlowControlMixin,
 protocols.SubprocessProtocol):
 """Like StreamReaderProtocol, but for a subprocess."""

 def __init__(self, limit, loop):
 super().__init__(loop=loop)
 self._limit = limit
 self.stdin = self.stdout = self.stderr = None
 self._transport = None
 self._process_exited = False
 self._pipe_fds = []
 self._stdin_closed = self._loop.create_future()

 def __repr__(self):
 info = [self.__class__.__name__]
 if self.stdin is not None:
 info.append(f'stdin={self.stdin!r}')
 if self.stdout is not None:
 info.append(f'stdout={self.stdout!r}')
 if self.stderr is not None:
 info.append(f'stderr={self.stderr!r}')
 return '<{}>'.format(' '.join(info))

 def connection_made(self, transport):
 self._transport = transport

 stdout_transport = transport.get_pipe_transport(1)
 if stdout_transport is not None:
 self.stdout = streams.StreamReader(limit=self._limit,
 loop=self._loop)
 self.stdout.set_transport(stdout_transport)
 self._pipe_fds.append(1)

 stderr_transport = transport.get_pipe_transport(2)
 if stderr_transport is not None:
 self.stderr = streams.StreamReader(limit=self._limit,
 loop=self._loop)
 self.stderr.set_transport(stderr_transport)
 self._pipe_fds.append(2)

 stdin_transport = transport.get_pipe_transport(0)
 if stdin_transport is not None:
 self.stdin = streams.StreamWriter(stdin_transport,
 protocol=self,
 reader=None,
 loop=self._loop)

 def pipe_data_received(self, fd, data):
 if fd == 1:
 reader = self.stdout
 elif fd == 2:
 reader = self.stderr
 else:
 reader = None
 if reader is not None:
 reader.feed_data(data)

 def pipe_connection_lost(self, fd, exc):
 if fd == 0:
 pipe = self.stdin
 if pipe is not None:
 pipe.close()
 self.connection_lost(exc)
 if exc is None:
 self._stdin_closed.set_result(None)
 else:
 self._stdin_closed.set_exception(exc)
 return
 if fd == 1:
 reader = self.stdout
 elif fd == 2:
 reader = self.stderr
 else:
 reader = None
 if reader is not None:
 if exc is None:
 reader.feed_eof()
 else:
 reader.set_exception(exc)

 if fd in self._pipe_fds:
 self._pipe_fds.remove(fd)
 self._maybe_close_transport()

 def process_exited(self):
 self._process_exited = True
 self._maybe_close_transport()

 def _maybe_close_transport(self):
 if len(self._pipe_fds) == 0 and self._process_exited:
 self._transport.close()
 self._transport = None

 def _get_close_waiter(self, stream):
 if stream is self.stdin:
 return self._stdin_closed


class Process:
 def __init__(self, transport, protocol, loop):
 self._transport = transport
 self._protocol = protocol
 self._loop = loop
 self.stdin = protocol.stdin
 self.stdout = protocol.stdout
 self.stderr = protocol.stderr
 self.pid = transport.get_pid()

 def __repr__(self):
 return f'<{self.__class__.__name__}{self.pid}>'

 @property
 def returncode(self):
 return self._transport.get_returncode()

 async def wait(self):
 """Wait until the process exit and return the process return code."""
 return await self._transport._wait()

 def send_signal(self, signal):
 self._transport.send_signal(signal)

 def terminate(self):
 self._transport.terminate()

 def kill(self):
 self._transport.kill()

 async def _feed_stdin(self, input):
 debug = self._loop.get_debug()
 self.stdin.write(input)
 if debug:
 logger.debug(
 '%r communicate: feed stdin (%s bytes)', self, len(input))
 try:
 await self.stdin.drain()
 except (BrokenPipeError, ConnectionResetError) as exc:
 # communicate() ignores BrokenPipeError and ConnectionResetError
 if debug:
 logger.debug('%r communicate: stdin got %r', self, exc)

 if debug:
 logger.debug('%r communicate: close stdin', self)
 self.stdin.close()

 async def _noop(self):
 return None

 async def _read_stream(self, fd):
 transport = self._transport.get_pipe_transport(fd)
 if fd == 2:
 stream = self.stderr
 else:
 assert fd == 1
 stream = self.stdout
 if self._loop.get_debug():
 name = 'stdout' if fd == 1 else 'stderr'
 logger.debug('%r communicate: read %s', self, name)
 output = await stream.read()
 if self._loop.get_debug():
 name = 'stdout' if fd == 1 else 'stderr'
 logger.debug('%r communicate: close %s', self, name)
 transport.close()
 return output

 async def communicate(self, input=None):
 if input is not None:
 stdin = self._feed_stdin(input)
 else:
 stdin = self._noop()
 if self.stdout is not None:
 stdout = self._read_stream(1)
 else:
 stdout = self._noop()
 if self.stderr is not None:
 stderr = self._read_stream(2)
 else:
 stderr = self._noop()
 stdin, stdout, stderr = await tasks.gather(stdin, stdout, stderr,
 loop=self._loop)
 await self.wait()
 return (stdout, stderr)


async def create_subprocess_shell(cmd, stdin=None, stdout=None, stderr=None,
 loop=None, limit=streams._DEFAULT_LIMIT,
 **kwds):
 if loop is None:
 loop = events.get_event_loop()
 else:
 warnings.warn("The loop argument is deprecated since Python 3.8 "
 "and scheduled for removal in Python 3.10.",
 DeprecationWarning,
 stacklevel=2
 )

 protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
 loop=loop)
 transport, protocol = await loop.subprocess_shell(
 protocol_factory,
 cmd, stdin=stdin, stdout=stdout,
 stderr=stderr, **kwds)
 return Process(transport, protocol, loop)


async def create_subprocess_exec(program, *args, stdin=None, stdout=None,
 stderr=None, loop=None,
 limit=streams._DEFAULT_LIMIT, **kwds):
 if loop is None:
 loop = events.get_event_loop()
 else:
 warnings.warn("The loop argument is deprecated since Python 3.8 "
 "and scheduled for removal in Python 3.10.",
 DeprecationWarning,
 stacklevel=2
 )
 protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
 loop=loop)
 transport, protocol = await loop.subprocess_exec(
 protocol_factory,
 program, *args,
 stdin=stdin, stdout=stdout,
 stderr=stderr, **kwds)
 return Process(transport, protocol, loop)