import abc
import collections.abc
import io
import typing

import trio.abc



T    = typing.TypeVar("T")
T_co = typing.TypeVar("T_co", covariant=True)
U_co = typing.TypeVar("U_co", covariant=True)


ArbitraryReceiveChannel = typing.Union[
	trio.abc.ReceiveChannel[T_co],
	typing.AsyncIterable[T_co],
	typing.Awaitable[T_co],
	typing.Iterable[T_co]
]


ArbitraryReceiveStream = typing.Union[
	trio.abc.ReceiveStream,
	typing.AsyncIterable[bytes],
	typing.Awaitable[bytes],
	typing.Iterable[bytes],
	bytes
]


class _ChannelSharedBase:
	__slots__ = ("lock", "refcount")
	
	lock:     trio.Lock
	refcount: int
	
	def __init__(self):
		self.lock     = trio.Lock()
		self.refcount = 1


class ReceiveChannel(trio.abc.ReceiveChannel[T_co], typing.Generic[T_co]):
	"""A slightly extended version of `trio`'s standard interface for receiving object streams.
	
	Attributes
	----------
	count
		The number of objects that will be returned, or `None` if unavailable
	atime
		Time of the entry's last access (before the current one) in seconds
		since the Unix epoch, or `None` if unkown
	mtime
		Time of the entry's last modification in seconds since the Unix epoch,
		or `None` if unknown
	btime
		Time of entry creation in seconds since the Unix epoch, or `None`
		if unknown
	"""
	__slots__ = ("count", "atime", "mtime", "btime")
	
	# The total length of this stream (if available)
	count: typing.Optional[int]
	
	# The backing record's last access time
	atime: typing.Optional[typing.Union[int, float]]
	# The backing record's last modification time
	mtime: typing.Optional[typing.Union[int, float]]
	# The backing record's creation (“birth”) time
	btime: typing.Optional[typing.Union[int, float]]
	

	def __init__(self):
		self.count = None
		self.atime = None
		self.mtime = None
		self.btime = None

	
	async def collect(self) -> typing.List[T_co]:
		result: typing.List[T_co] = []
		async with self:
			async for item in self:
				result.append(item)
		return result


class _TeeingChannelShared(_ChannelSharedBase, typing.Generic[T_co]):
	__slots__ = ("nursery", "nursery_manager", "channels", "bufsize", "source")
	
	nursery:         trio._core._run.Nursery  # No public type for this
	nursery_manager: trio._core._run.NurseryManager  # No public type for this
	
	channels: typing.List[trio.abc.SendChannel[T_co]]
	
	bufsize: int
	source:  typing.Optional[trio.abc.ReceiveChannel[T_co]]


class TeeingReceiveChannel(ReceiveChannel[T_co], typing.Generic[T_co]):
	"""Allows the value retrieved from a single `trio.abc.ReceiveChannel` to be
	pushed towards several receivers additionally to be received by the caller
	it is returned to.
	"""
	
	__slots__ = ("_closed", "_shared")
	
	_closed: bool
	_shared: _TeeingChannelShared[T_co]
	
	def __init__(self, source: trio.abc.ReceiveChannel[T_co], buffer_size: int = 0, *,
	             _shared: typing.Optional[_TeeingChannelShared[T_co]] = None):
		super().__init__()
		
		if _shared is not None:
			self._shared = _shared
			return
		
		self._shared = _TeeingChannelShared() if _shared is None else _shared
		self._shared.bufsize = buffer_size
		self._shared.source  = source
		
		# Try to copy extra attributes from source channel
		self.count = getattr(source, "count", None)
		self.atime = getattr(source, "atime", None)
		self.mtime = getattr(source, "mtime", None)
		self.btime = getattr(source, "btime", None)
		
		self._closed = False
		
		# Create nursery without using a `async with`-statement
		# (Only works because the `__aenter__`-call does not actually block on anything.)
		self._shared.channels = []
		self._shared.nursery_manager = trio.open_nursery()
		try:
			self._shared.nursery_manager.__aenter__().__await__().send(None)
		except StopIteration as exc:
			self._shared.nursery = exc.args[0]
		else:
			raise RuntimeError("Failed to initialize nursery synchronously")
	
	
	async def start_task(self, func: typing.Callable[[trio.abc.ReceiveChannel[T_co]], T], *args) -> T:
		async with self._shared.lock:  # type: ignore[attr-defined] # upstream type bug # noqa: F821
			send_channel, receive_channel = trio.open_memory_channel(self._shared.bufsize)
			result = await self._shared.nursery.start(func, receive_channel, *args)
			self._shared.channels.append(send_channel)
		return result
	
	
	def start_task_soon(
			self, func: typing.Callable[[trio.abc.ReceiveChannel[T_co]], typing.Any], *args
	) -> None:
		# Doing this sync is just wrong, but we cannot block in this function…
		self._shared.lock.acquire_nowait()
		
		try:
			send_channel, receive_channel = trio.open_memory_channel(self._shared.bufsize)
			self._shared.nursery.start_soon(func, receive_channel, *args)
			self._shared.channels.append(send_channel)
		finally:
			self._shared.lock.release()
	
	
	def clone(self) -> ReceiveChannel[T_co]:
		if self._closed:
			raise trio.ClosedResourceError()
		
		try:
			# Cast source value to ignore the possible `None` variant as the
			# passed source value will be ignored if we provide `_shared`
			source = typing.cast(trio.abc.ReceiveChannel[T_co], self._shared.source)
			
			return TeeingReceiveChannel(source, _shared=self._shared)
		except BaseException:
			raise
		else:
			self._shared.refcount += 1
	
	
	async def receive(self) -> T_co:
		if self._closed:
			raise trio.ClosedResourceError()
		if self._shared.source is None:
			raise trio.EndOfChannel()
		
		try:
			# Pass received value (or EOF) to waiting write tasks
			async with self._shared.lock:  # type: ignore[attr-defined] # upstream type bug # noqa: F821
				try:
					value = await self._shared.source.receive()
					for channel in self._shared.channels:
						await channel.send(value)
					return value
				except trio.EndOfChannel:
					for channel in self._shared.channels:
						await channel.aclose()
					self._shared.channels.clear()
					raise
		except trio.BrokenResourceError:
			await self.aclose(_mark_closed=True)
			raise
		except trio.EndOfChannel:
			# Ensure that our slaves have finished before the final value is
			# returned
			await self.aclose(_mark_closed=False)
			raise
	
	
	def receive_nowait(self) -> T_co:
		if self._closed:
			raise trio.ClosedResourceError()
		if self._shared.source is None:
			raise trio.EndOfChannel()
		
		# We implement this as there is no guarantee that *all* of our teeing
		# channels will accept the received value non-blockingly
		raise trio.WouldBlock()
	
	
	async def aclose(self, *, _mark_closed=True) -> None:
		if _mark_closed:
			self._closed = True
		
		if self._shared.source is None:
			return
		
		self._shared.refcount -= 1
		if self._shared.refcount != 0:
			return
		
		try:
			try:
				# Close all remaining teeing streams by sending them a
				# cancellation error
				for channel in self._shared.channels:
					with trio.CancelScope() as cancel_scope:
						cancel_scope.shield = True
						await trio.aclose_forcefully(channel)
				
				# Close the source stream
				await self._shared.source.aclose()
			except BaseException as exc:
				# Wait for any tasks possibly still active in the nursery
				#
				# This must be called in this special `try` way to ensure that
				# the nursery will be properly closed down even if its contents
				# cannot be cleaned up anymore. Additionally, this will replace
				# a `trio.Cancelled` resulting of some other temporarily stored
				# exception by the actual exception value originally raised.
				if not await self._shared.nursery_manager.__aexit__(type(exc), exc, exc.__traceback__):
					raise
			else:
				await self._shared.nursery_manager.__aexit__(None, None, None)
		finally:
			self._shared.channels.clear()
			self._shared.source = None



class _WrapingTrioReceiveChannel(ReceiveChannel[T_co], typing.Generic[T_co]):
	__slots__ = ("_source",)
	
	_source: trio.abc.ReceiveChannel[T_co]
	
	
	def __init__(self, source: trio.abc.ReceiveChannel[T_co]):
		super().__init__()
		
		self._source = source
	
	
	async def receive(self) -> T_co:
		return await self._source.receive()
	
	
	def receive_nowait(self) -> T_co:
		return self._source.receive_nowait()
	
	
	def clone(self) -> ReceiveChannel[T_co]:
		return self.__class__(self._source.clone())
	
	
	async def aclose(self) -> None:
		await self._source.aclose()



class _WrapingChannelShared(_ChannelSharedBase, typing.Generic[U_co]):
	__slots__ = ("source",)
	
	source: typing.Optional[U_co]
	
	
	def __init__(self, source: U_co):
		super().__init__()
		
		self.source = source



class _WrapingIterReceiveChannelBase(ReceiveChannel[T_co], typing.Generic[T_co, U_co]):
	"""Abstracts over various forms of synchronous and asynchronous returning of
	   object streams
	"""
	__slots__ = ("_closed", "_shared")
	
	_closed: bool
	_shared: _WrapingChannelShared[U_co]
	
	def __init__(self, source: typing.Optional[U_co], *,
	             _shared: typing.Optional[_WrapingChannelShared[U_co]] = None):
		super().__init__()
		
		assert source is not None or _shared is not None
		
		self._closed = False
		if _shared is None:
			assert source is not None
			self._shared = _WrapingChannelShared(source)
		else:
			self._shared = _shared
	
	
	@abc.abstractmethod
	async def _receive(self) -> T_co:
		pass
	
	
	async def receive(self) -> T_co:
		if self._closed:
			raise trio.ClosedResourceError()
		if self._shared.source is None:
			raise trio.EndOfChannel()
		
		try:
			async with self._shared.lock:  # type: ignore[attr-defined]  # upstream type bug # noqa: F821
				return await self._receive()
		except trio.BrokenResourceError:
			await self.aclose(_mark_closed=True)
			raise
		except trio.EndOfChannel:
			await self.aclose(_mark_closed=False)
			raise
	
	
	@abc.abstractmethod
	def _receive_nowait(self) -> T_co:
		pass
	
	
	def receive_nowait(self) -> T_co:
		if self._closed:
			raise trio.ClosedResourceError()
		if self._shared.source is None:
			raise trio.EndOfChannel()
		
		self._shared.lock.acquire_nowait()
		try:
			return self._receive_nowait()
		finally:
			self._shared.lock.release()
	
	
	def clone(self) -> ReceiveChannel[T_co]:
		if self._closed:
			raise trio.ClosedResourceError()
		
		try:
			return self.__class__(None, _shared=self._shared)
		except BaseException:
			raise
		else:
			self._shared.refcount += 1
	
	
	@abc.abstractmethod
	async def _close_source(self) -> None:
		pass
	
	
	async def aclose(self, *, _mark_closed: bool = True) -> None:
		if not self._closed and _mark_closed:
			self._closed = True
		
		if self._shared.source is None:
			return
		
		self._shared.refcount -= 1
		if self._shared.refcount != 0:
			return
		
		try:
			await self._close_source()
		except AttributeError:
			pass
		finally:
			self._shared.source = None


class _WrapingAsyncIterReceiveChannel(
		_WrapingIterReceiveChannelBase[T_co, typing.AsyncIterator[T_co]],
		typing.Generic[T_co]
):
	def __init__(self, source: typing.Optional[typing.AsyncIterable[T_co]], **kwargs):
		super().__init__(source.__aiter__() if source is not None else None, **kwargs)
	
	
	async def _receive(self) -> T_co:
		assert self._shared.source is not None
		
		try:
			return await self._shared.source.__anext__()
		except StopAsyncIteration as exc:
			raise trio.EndOfChannel() from exc
	
	
	def _receive_nowait(self) -> T_co:
		# Cannot ask this stream type for a non-blocking value
		raise trio.WouldBlock()
	
	
	async def _close_source(self) -> None:
		try:
			await self._shared.source.aclose()  # type: ignore  # We catch errors instead
		except AttributeError:
			pass



class _WrapingSyncIterReceiveChannel(
		_WrapingIterReceiveChannelBase[T_co, typing.Iterator[T_co]],
		typing.Generic[T_co]
):
	def __init__(self, source: typing.Optional[typing.Iterable[T_co]],
	             count_hint: typing.Optional[int] = None, **kwargs):
		super().__init__(iter(source) if source is not None else None, **kwargs)
		
		self.count = count_hint
	
	
	async def _receive(self) -> T_co:
		assert self._shared.source is not None
		
		try:
			return next(self._shared.source)
		except StopIteration as exc:
			raise trio.EndOfChannel() from exc
	
	
	def _receive_nowait(self) -> T_co:
		assert self._shared.source is not None
		
		try:
			return next(self._shared.source)
		except StopIteration:
			# We cannot handle invoking async close here
			raise trio.WouldBlock() from None
	
	
	async def _close_source(self) -> None:
		try:
			self._shared.source.close()  # type: ignore  # We catch errors instead
		except AttributeError:
			pass



def receive_channel_from(channel: ArbitraryReceiveChannel[T_co]) -> ReceiveChannel[T_co]:
	# Optimization: Reuse given stream object, rather then creating a new
	#               wrapper when it is already of the right interface type
	if isinstance(channel, ReceiveChannel):
		return channel
	
	# Optimization: Wrap the given Trio stream object in a tiny wrapper that
	#               just passes through all calls, but adds our extra fields
	#               and values
	if isinstance(channel, trio.abc.ReceiveChannel):
		return _WrapingTrioReceiveChannel(channel)
	
	# Handle asynchronous iterables
	if isinstance(channel, (collections.abc.AsyncIterable, collections.abc.Awaitable)):
		source1: typing.AsyncIterable[T_co]
		
		if isinstance(channel, collections.abc.Awaitable):
			async def await_iter_wrapper(channel: typing.Awaitable[T_co]) \
			      -> typing.AsyncIterable[T_co]:
				yield await channel
			source1 = await_iter_wrapper(channel)
		else:
			source1 = channel
		
		return _WrapingAsyncIterReceiveChannel(source1)
	
	# Handle synchronous iterables (and try to deduce the length in each case possible)
	if isinstance(channel, collections.abc.Iterable):
		count:  typing.Optional[int]  = None
		source2: typing.Iterable[T_co] = channel
		
		if isinstance(source2, collections.abc.Sequence):
			count = len(source2)
		
		return _WrapingSyncIterReceiveChannel(source2, count)
	
	assert False, "Unreachable code"



class ReceiveStream(trio.abc.ReceiveStream):
	"""A slightly extended version of `trio`'s standard interface for receiving byte streams.
	
	Attributes
	----------
	size
		The size of the entire stream data in bytes, or `None` if unavailable
	atime
		Time of the entry's last access (before the current one) in seconds
		since the Unix epoch, or `None` if unkown
	mtime
		Time of the entry's last modification in seconds since the Unix epoch,
		or `None` if unknown
	btime
		Time of entry creation in seconds since the Unix epoch, or `None`
		if unknown
	"""
	__slots__ = ("size", "atime", "mtime", "btime")
	
	# The total length of this stream (if available)
	size: typing.Optional[int]
	
	# The backing record's last access time
	atime: typing.Optional[typing.Union[int, float]]
	# The backing record's last modification time
	mtime: typing.Optional[typing.Union[int, float]]
	# The backing record's creation (“birth”) time
	btime: typing.Optional[typing.Union[int, float]]
	
	
	def __init__(self):
		self.size  = None
		self.atime = None
		self.mtime = None
		self.btime = None
	
	
	async def collect(self) -> bytes:
		value = bytearray()
		async with self:
			# Use “size”, if available, to try and read the entire stream's conents
			# in one go
			max_bytes = getattr(self, "size", None)
			
			while True:
				chunk = await self.receive_some(max_bytes)
				if len(chunk) < 1:
					break
				value += chunk
		return bytes(value)


class TeeingReceiveStream(ReceiveStream):
	"""Allows the value retrieved from a single `trio.abc.ReceiveStream` to be
	pushed towards several receivers additionally to be received by the caller
	it is returned to.
	"""
	
	__slots__ = ("_nursery", "_nursery_manager", "_channels", "_bufsize", "_closed", "_source")
	
	_nursery:         trio._core._run.Nursery  # No public type for this
	_nursery_manager: trio._core._run.NurseryManager  # No public type for this
	
	_channels: typing.List[trio.abc.SendChannel[bytes]]
	
	_bufsize: int
	_closed:  bool
	_source:  typing.Optional[trio.abc.ReceiveStream]
	
	def __init__(self, source: trio.abc.ReceiveStream, buffer_size: int = 0):
		super().__init__()
		
		self._bufsize = buffer_size
		self._source  = source
		self._closed  = False
		
		# Try to copy extra attributes from source stream
		self.size  = getattr(source, "size", None)
		self.atime = getattr(source, "atime", None)
		self.mtime = getattr(source, "mtime", None)
		self.btime = getattr(source, "btime", None)
		
		# Create nursery without using a `async with`-statement
		# (Only works because the `__aenter__`-call does not actually block on anything.)
		self._channels = []
		self._nursery_manager = trio.open_nursery()
		try:
			self._nursery_manager.__aenter__().__await__().send(None)
		except StopIteration as exc:
			self._nursery = exc.args[0]
		else:
			raise RuntimeError("Failed to initialize nursery synchronously")
	
	
	async def start_task(self, func: typing.Callable[[trio.abc.ReceiveStream], T], *args) -> T:
		send_channel, receive_channel = trio.open_memory_channel(self._bufsize)
		result = await self._nursery.start(func, receive_stream_from(receive_channel), *args)
		self._channels.append(send_channel)
		return result
	
	
	def start_task_soon(
			self, func: typing.Callable[[trio.abc.ReceiveStream], typing.Any], *args
	) -> None:
		send_channel, receive_channel = trio.open_memory_channel(self._bufsize)
		self._nursery.start_soon(func, receive_stream_from(receive_channel), *args)
		self._channels.append(send_channel)
	
	
	async def receive_some(self, max_bytes: typing.Optional[int] = None) -> bytes:
		if self._closed:
			raise trio.ClosedResourceError()
		if self._source is None:
			return b""
		
		try:
			value = await self._source.receive_some(max_bytes)
			
			# Pass received value (or EOF) to waiting write tasks
			if len(value) > 0:
				for channel in self._channels:
					await channel.send(value)
			else:
				for channel in self._channels:
					await channel.aclose()
				self._channels.clear()

				# Ensure that our slaves have finished before the final value
				# is returned
				await self.aclose(_mark_closed=False)
			
			return value
		except BaseException:
			await self.aclose()
			raise
	
	
	async def aclose(self, *, _mark_closed=True) -> None:
		if _mark_closed:
			self._closed = True
		if self._source is None:
			return
		
		try:
			try:
				# Close all remaining teeing streams by sending them a
				# cancellation error
				for channel in self._channels:
					with trio.CancelScope() as cancel_scope:
						cancel_scope.shield = True
						await trio.aclose_forcefully(channel)
				
				# Close the source stream
				await self._source.aclose()
			except BaseException as exc:
				# Wait for any tasks possibly still active in the nursery
				#
				# This must be called in this special `try` way to ensure that
				# the nursery will be properly closed down even if its contents
				# cannot be cleaned up anymore. Additionally, this will replace
				# a `trio.Cancelled` resulting of some other temporarily stored
				# exception by the actual exception value originally raised.
				if not await self._nursery_manager.__aexit__(type(exc), exc, exc.__traceback__):
					raise
			else:
				await self._nursery_manager.__aexit__(None, None, None)
		finally:
			self._channels.clear()
			self._source = None


class _WrapingTrioReceiveStream(ReceiveStream):
	"""
	Abstracts over a bare `trio.abc.ReceiveStream` to add our standard fields
	and methods to that stream
	"""
	__slots__ = ("_source",)
	
	_source: trio.abc.ReceiveStream
	
	def __init__(self, source: trio.abc.ReceiveStream):
		super().__init__()
		
		self._source = source
	
	
	async def receive_some(self, max_bytes: typing.Optional[int] = None) -> bytes:
		return await self._source.receive_some(max_bytes)
	
	
	async def aclose(self) -> None:
		await self._source.aclose()



class _WrapingIterReceiveStreamBase(ReceiveStream, typing.Generic[T_co]):
	"""Abstracts over various forms of synchronous and asynchronous returning of
	   byte streams
	"""
	
	__slots__ = ("_buffer", "_memview", "_offset", "_closed", "_source")
	
	_buffer:  bytearray
	_memview: typing.Optional[memoryview]
	_offset:  int
	_closed:  bool
	
	_source: typing.Optional[T_co]
	
	def __init__(self, source: T_co):
		super().__init__()
		
		self._source = source
		
		self._buffer  = bytearray()
		self._memview = None
		self._offset  = 0
		self._closed  = False
	
	
	@abc.abstractmethod
	async def _receive(self, max_bytes: typing.Optional[int]) -> bytes:
		pass
	
	
	async def receive_some(self, max_bytes: typing.Optional[int] = None) -> bytes:
		if self._closed:
			raise trio.ClosedResourceError()
		if self._source is None:
			return b""
		
		# Serve chunks from buffer if there is any data that hasn't been
		# delivered yet
		if self._memview:
			if max_bytes is not None:
				end_offset = min(self._offset + max_bytes, len(self._memview))
			else:
				end_offset = len(self._memview)
			
			result = bytes(self._memview[self._offset:end_offset])
			if end_offset >= len(self._memview):
				self._offset = 0
				self._memview.release()  # type: ignore  # Fixed in mypy 0.350
				self._memview = None
				self._buffer.clear()
			return result
		
		
		value = await self._receive(max_bytes)
		
		assert isinstance(value, bytes), \
		       f"Source stream {repr(self._source)} returned non-byte segment"
		
		if len(value) < 1:
			# We're at the end
			await self.aclose(_mark_closed=False)
			return b""
		
		# Stash extra bytes that are too large for our receiver
		if max_bytes is not None and max_bytes > len(value):
			self._buffer += value[max_bytes:]
			self._memview = memoryview(self._buffer)
			value = value[:max_bytes]
		
		return value
	
	
	@abc.abstractmethod
	async def _close_source(self) -> None:
		pass
	
	
	async def aclose(self, *, _mark_closed=True) -> None:
		if not self._closed and _mark_closed:
			self._closed = True
		
		if self._source is None:
			return
		
		try:
			await self._close_source()
		finally:
			self._source = None
			if self._memview is not None:
				self._memview.release()  # type: ignore  # Fixed in mypy 0.450
				self._memview = None
			self._buffer.clear()


class _WrapingAsyncIterReceiveStream(_WrapingIterReceiveStreamBase[typing.AsyncIterator[bytes]]):
	def __init__(self, source: typing.AsyncIterable[bytes]):
		super().__init__(source.__aiter__())
	
	
	async def _receive(self, _: typing.Optional[int]) -> bytes:
		assert self._source is not None
		
		# Skip empty returned byte strings as they have a special meaning here
		value = b""
		while len(value) < 1:
			try:
				value = await self._source.__anext__()
			except StopAsyncIteration:
				return b""
		return value
	
	
	async def _close_source(self) -> None:
		try:
			await self._source.aclose()  # type: ignore  # We catch errors instead
		except AttributeError:
			pass



class _WrapingSyncIterReceiveStream(_WrapingIterReceiveStreamBase[typing.Iterator[bytes]]):
	def __init__(self, source: typing.Iterable[bytes], size_hint: typing.Optional[int]):
		super().__init__(iter(source))
		
		self.size = size_hint
	
	
	async def _receive(self, _: typing.Optional[int]) -> bytes:
		assert self._source is not None
		
		# Skip empty returned byte strings as they have a special meaning here
		value = b""
		while len(value) < 1:
			try:
				value = next(self._source)
			except StopIteration:
				return b""
		return value
	
	
	async def _close_source(self) -> None:
		try:
			# We catch errors instead
			self._source.close()  # type: ignore[union-attr] # noqa: F821
		except AttributeError:
			pass



def receive_stream_from(stream: ArbitraryReceiveStream) -> ReceiveStream:
	# Optimization: Reuse given stream object, rather then creating a new
	#               wrapper when it is already of the right interface type
	if isinstance(stream, ReceiveStream):
		return stream
	
	# Optimization: Wrap the given Trio stream object in a tiny wrapper that
	#               just passes through all calls, but adds our extra fields
	#               and values
	if isinstance(stream, trio.abc.ReceiveStream):
		return _WrapingTrioReceiveStream(stream)
	
	# Handle asynchronous iterables
	if isinstance(stream, (collections.abc.AsyncIterable, collections.abc.Awaitable)):
		source1: typing.AsyncIterable[bytes]
		
		# Wrap awaitables of bytes in an asynchronous iterable that yields once
		if isinstance(stream, collections.abc.Awaitable):
			async def await_iter_wrapper(stream: typing.Awaitable[bytes]) \
			      -> typing.AsyncIterable[bytes]:
				yield await stream
			source1 = await_iter_wrapper(stream)
		else:
			source1 = stream
		
		return _WrapingAsyncIterReceiveStream(source1)
	
	# Handle synchronous iterables (and try to deduce the length in each case possible)
	if isinstance(stream, (bytes, collections.abc.Iterable)):
		size:    typing.Optional[int] = None
		source2: typing.Iterable[bytes]
		
		# Wrap a simple bytes value in a tuple to make it an iterable
		if isinstance(stream, bytes):
			source2 = (stream,)
		else:
			source2 = stream
		
		# Deduce the length of sequences of bytes (including the single bytes sequence above)
		if isinstance(source2, collections.abc.Sequence):
			size = sum(len(item) for item in source2)
		# … and in-memory byte stream files as well
		elif isinstance(source2, io.BytesIO):
			# Asks for just the remaining length, restoring the position afterwards
			pos = source2.tell()
			source2.seek(0, io.SEEK_END)
			size = source2.tell() - pos
			source2.seek(pos, io.SEEK_SET)
		
		return _WrapingSyncIterReceiveStream(source2, size)
	
	assert False, "Unreachable code"

import abc
import collections.abc
import io
import typing

import trio.abc



T    = typing.TypeVar("T")
T_co = typing.TypeVar("T_co", covariant=True)
U_co = typing.TypeVar("U_co", covariant=True)


ArbitraryReceiveChannel = typing.Union[
	trio.abc.ReceiveChannel[T_co],
	typing.AsyncIterable[T_co],
	typing.Awaitable[T_co],
	typing.Iterable[T_co]
]


ArbitraryReceiveStream = typing.Union[
	trio.abc.ReceiveStream,
	typing.AsyncIterable[bytes],
	typing.Awaitable[bytes],
	typing.Iterable[bytes],
	bytes
]


class _ChannelSharedBase:
	__slots__ = ("lock", "refcount")
	
	lock:     trio.Lock
	refcount: int
	
	def __init__(self):
		self.lock     = trio.Lock()
		self.refcount = 1


class ReceiveChannel(trio.abc.ReceiveChannel[T_co], typing.Generic[T_co]):
	"""A slightly extended version of `trio`'s standard interface for receiving object streams.
	
	Attributes
	----------
	count
		The number of objects that will be returned, or `None` if unavailable
	atime
		Time of the entry's last access (before the current one) in seconds
		since the Unix epoch, or `None` if unkown
	mtime
		Time of the entry's last modification in seconds since the Unix epoch,
		or `None` if unknown
	btime
		Time of entry creation in seconds since the Unix epoch, or `None`
		if unknown
	"""
	__slots__ = ("count", "atime", "mtime", "btime")
	
	# The total length of this stream (if available)
	count: typing.Optional[int]
	
	# The backing record's last access time
	atime: typing.Optional[typing.Union[int, float]]
	# The backing record's last modification time
	mtime: typing.Optional[typing.Union[int, float]]
	# The backing record's creation (“birth”) time
	btime: typing.Optional[typing.Union[int, float]]
	

	def __init__(self):
		self.count = None
		self.atime = None
		self.mtime = None
		self.btime = None

	
	async def collect(self) -> typing.List[T_co]:
		result: typing.List[T_co] = []
		async with self:
			async for item in self:
				result.append(item)
		return result


class _TeeingChannelShared(_ChannelSharedBase, typing.Generic[T_co]):
	__slots__ = ("nursery", "nursery_manager", "channels", "bufsize", "source")
	
	nursery:         trio._core._run.Nursery  # No public type for this
	nursery_manager: trio._core._run.NurseryManager  # No public type for this
	
	channels: typing.List[trio.abc.SendChannel[T_co]]
	
	bufsize: int
	source:  typing.Optional[trio.abc.ReceiveChannel[T_co]]


class TeeingReceiveChannel(ReceiveChannel[T_co], typing.Generic[T_co]):
	"""Allows the value retrieved from a single `trio.abc.ReceiveChannel` to be
	pushed towards several receivers additionally to be received by the caller
	it is returned to.
	"""
	
	__slots__ = ("_closed", "_shared")
	
	_closed: bool
	_shared: _TeeingChannelShared[T_co]
	
	def __init__(self, source: trio.abc.ReceiveChannel[T_co], buffer_size: int = 0, *,
	             _shared: typing.Optional[_TeeingChannelShared[T_co]] = None):
		super().__init__()
		
		if _shared is not None:
			self._shared = _shared
			return
		
		self._shared = _TeeingChannelShared() if _shared is None else _shared
		self._shared.bufsize = buffer_size
		self._shared.source  = source
		
		# Try to copy extra attributes from source channel
		self.count = getattr(source, "count", None)
		self.atime = getattr(source, "atime", None)
		self.mtime = getattr(source, "mtime", None)
		self.btime = getattr(source, "btime", None)
		
		self._closed = False
		
		# Create nursery without using a `async with`-statement
		# (Only works because the `__aenter__`-call does not actually block on anything.)
		self._shared.channels = []
		self._shared.nursery_manager = trio.open_nursery()
		try:
			self._shared.nursery_manager.__aenter__().__await__().send(None)
		except StopIteration as exc:
			self._shared.nursery = exc.args[0]
		else:
			raise RuntimeError("Failed to initialize nursery synchronously")
	
	
	async def start_task(self, func: typing.Callable[[trio.abc.ReceiveChannel[T_co]], T], *args) -> T:
		async with self._shared.lock:  # type: ignore[attr-defined] # upstream type bug # noqa: F821
			send_channel, receive_channel = trio.open_memory_channel(self._shared.bufsize)
			result = await self._shared.nursery.start(func, receive_channel, *args)
			self._shared.channels.append(send_channel)
		return result
	
	
	def start_task_soon(
			self, func: typing.Callable[[trio.abc.ReceiveChannel[T_co]], typing.Any], *args
	) -> None:
		# Doing this sync is just wrong, but we cannot block in this function…
		self._shared.lock.acquire_nowait()
		
		try:
			send_channel, receive_channel = trio.open_memory_channel(self._shared.bufsize)
			self._shared.nursery.start_soon(func, receive_channel, *args)
			self._shared.channels.append(send_channel)
		finally:
			self._shared.lock.release()
	
	
	def clone(self) -> ReceiveChannel[T_co]:
		if self._closed:
			raise trio.ClosedResourceError()
		
		try:
			# Cast source value to ignore the possible `None` variant as the
			# passed source value will be ignored if we provide `_shared`
			source = typing.cast(trio.abc.ReceiveChannel[T_co], self._shared.source)
			
			return TeeingReceiveChannel(source, _shared=self._shared)
		except BaseException:
			raise
		else:
			self._shared.refcount += 1
	
	
	async def receive(self) -> T_co:
		if self._closed:
			raise trio.ClosedResourceError()
		if self._shared.source is None:
			raise trio.EndOfChannel()
		
		try:
			# Pass received value (or EOF) to waiting write tasks
			async with self._shared.lock:  # type: ignore[attr-defined] # upstream type bug # noqa: F821
				try:
					value = await self._shared.source.receive()
					for channel in self._shared.channels:
						await channel.send(value)
					return value
				except trio.EndOfChannel:
					for channel in self._shared.channels:
						await channel.aclose()
					self._shared.channels.clear()
					raise
		except trio.BrokenResourceError:
			await self.aclose(_mark_closed=True)
			raise
		except trio.EndOfChannel:
			# Ensure that our slaves have finished before the final value is
			# returned
			await self.aclose(_mark_closed=False)
			raise
	
	
	def receive_nowait(self) -> T_co:
		if self._closed:
			raise trio.ClosedResourceError()
		if self._shared.source is None:
			raise trio.EndOfChannel()
		
		# We implement this as there is no guarantee that *all* of our teeing
		# channels will accept the received value non-blockingly
		raise trio.WouldBlock()
	
	
	async def aclose(self, *, _mark_closed=True) -> None:
		if _mark_closed:
			self._closed = True
		
		if self._shared.source is None:
			return
		
		self._shared.refcount -= 1
		if self._shared.refcount != 0:
			return
		
		try:
			try:
				# Close all remaining teeing streams by sending them a
				# cancellation error
				for channel in self._shared.channels:
					with trio.CancelScope() as cancel_scope:
						cancel_scope.shield = True
						await trio.aclose_forcefully(channel)
				
				# Close the source stream
				await self._shared.source.aclose()
			except BaseException as exc:
				# Wait for any tasks possibly still active in the nursery
				#
				# This must be called in this special `try` way to ensure that
				# the nursery will be properly closed down even if its contents
				# cannot be cleaned up anymore. Additionally, this will replace
				# a `trio.Cancelled` resulting of some other temporarily stored
				# exception by the actual exception value originally raised.
				if not await self._shared.nursery_manager.__aexit__(type(exc), exc, exc.__traceback__):
					raise
			else:
				await self._shared.nursery_manager.__aexit__(None, None, None)
		finally:
			self._shared.channels.clear()
			self._shared.source = None



class _WrapingTrioReceiveChannel(ReceiveChannel[T_co], typing.Generic[T_co]):
	__slots__ = ("_source",)
	
	_source: trio.abc.ReceiveChannel[T_co]
	
	
	def __init__(self, source: trio.abc.ReceiveChannel[T_co]):
		super().__init__()
		
		self._source = source
	
	
	async def receive(self) -> T_co:
		return await self._source.receive()
	
	
	def receive_nowait(self) -> T_co:
		return self._source.receive_nowait()
	
	
	def clone(self) -> ReceiveChannel[T_co]:
		return self.__class__(self._source.clone())
	
	
	async def aclose(self) -> None:
		await self._source.aclose()



class _WrapingChannelShared(_ChannelSharedBase, typing.Generic[U_co]):
	__slots__ = ("source",)
	
	source: typing.Optional[U_co]
	
	
	def __init__(self, source: U_co):
		super().__init__()
		
		self.source = source



class _WrapingIterReceiveChannelBase(ReceiveChannel[T_co], typing.Generic[T_co, U_co]):
	"""Abstracts over various forms of synchronous and asynchronous returning of
	   object streams
	"""
	__slots__ = ("_closed", "_shared")
	
	_closed: bool
	_shared: _WrapingChannelShared[U_co]
	
	def __init__(self, source: typing.Optional[U_co], *,
	             _shared: typing.Optional[_WrapingChannelShared[U_co]] = None):
		super().__init__()
		
		assert source is not None or _shared is not None
		
		self._closed = False
		if _shared is None:
			assert source is not None
			self._shared = _WrapingChannelShared(source)
		else:
			self._shared = _shared
	
	
	@abc.abstractmethod
	async def _receive(self) -> T_co:
		pass
	
	
	async def receive(self) -> T_co:
		if self._closed:
			raise trio.ClosedResourceError()
		if self._shared.source is None:
			raise trio.EndOfChannel()
		
		try:
			async with self._shared.lock:  # type: ignore[attr-defined]  # upstream type bug # noqa: F821
				return await self._receive()
		except trio.BrokenResourceError:
			await self.aclose(_mark_closed=True)
			raise
		except trio.EndOfChannel:
			await self.aclose(_mark_closed=False)
			raise
	
	
	@abc.abstractmethod
	def _receive_nowait(self) -> T_co:
		pass
	
	
	def receive_nowait(self) -> T_co:
		if self._closed:
			raise trio.ClosedResourceError()
		if self._shared.source is None:
			raise trio.EndOfChannel()
		
		self._shared.lock.acquire_nowait()
		try:
			return self._receive_nowait()
		finally:
			self._shared.lock.release()
	
	
	def clone(self) -> ReceiveChannel[T_co]:
		if self._closed:
			raise trio.ClosedResourceError()
		
		try:
			return self.__class__(None, _shared=self._shared)
		except BaseException:
			raise
		else:
			self._shared.refcount += 1
	
	
	@abc.abstractmethod
	async def _close_source(self) -> None:
		pass
	
	
	async def aclose(self, *, _mark_closed: bool = True) -> None:
		if not self._closed and _mark_closed:
			self._closed = True
		
		if self._shared.source is None:
			return
		
		self._shared.refcount -= 1
		if self._shared.refcount != 0:
			return
		
		try:
			await self._close_source()
		except AttributeError:
			pass
		finally:
			self._shared.source = None


class _WrapingAsyncIterReceiveChannel(
		_WrapingIterReceiveChannelBase[T_co, typing.AsyncIterator[T_co]],
		typing.Generic[T_co]
):
	def __init__(self, source: typing.Optional[typing.AsyncIterable[T_co]], **kwargs):
		super().__init__(source.__aiter__() if source is not None else None, **kwargs)
	
	
	async def _receive(self) -> T_co:
		assert self._shared.source is not None
		
		try:
			return await self._shared.source.__anext__()
		except StopAsyncIteration as exc:
			raise trio.EndOfChannel() from exc
	
	
	def _receive_nowait(self) -> T_co:
		# Cannot ask this stream type for a non-blocking value
		raise trio.WouldBlock()
	
	
	async def _close_source(self) -> None:
		try:
			await self._shared.source.aclose()  # type: ignore  # We catch errors instead
		except AttributeError:
			pass



class _WrapingSyncIterReceiveChannel(
		_WrapingIterReceiveChannelBase[T_co, typing.Iterator[T_co]],
		typing.Generic[T_co]
):
	def __init__(self, source: typing.Optional[typing.Iterable[T_co]],
	             count_hint: typing.Optional[int] = None, **kwargs):
		super().__init__(iter(source) if source is not None else None, **kwargs)
		
		self.count = count_hint
	
	
	async def _receive(self) -> T_co:
		assert self._shared.source is not None
		
		try:
			return next(self._shared.source)
		except StopIteration as exc:
			raise trio.EndOfChannel() from exc
	
	
	def _receive_nowait(self) -> T_co:
		assert self._shared.source is not None
		
		try:
			return next(self._shared.source)
		except StopIteration:
			# We cannot handle invoking async close here
			raise trio.WouldBlock() from None
	
	
	async def _close_source(self) -> None:
		try:
			self._shared.source.close()  # type: ignore  # We catch errors instead
		except AttributeError:
			pass



def receive_channel_from(channel: ArbitraryReceiveChannel[T_co]) -> ReceiveChannel[T_co]:
	# Optimization: Reuse given stream object, rather then creating a new
	#               wrapper when it is already of the right interface type
	if isinstance(channel, ReceiveChannel):
		return channel
	
	# Optimization: Wrap the given Trio stream object in a tiny wrapper that
	#               just passes through all calls, but adds our extra fields
	#               and values
	if isinstance(channel, trio.abc.ReceiveChannel):
		return _WrapingTrioReceiveChannel(channel)
	
	# Handle asynchronous iterables
	if isinstance(channel, (collections.abc.AsyncIterable, collections.abc.Awaitable)):
		source1: typing.AsyncIterable[T_co]
		
		if isinstance(channel, collections.abc.Awaitable):
			async def await_iter_wrapper(channel: typing.Awaitable[T_co]) \
			      -> typing.AsyncIterable[T_co]:
				yield await channel
			source1 = await_iter_wrapper(channel)
		else:
			source1 = channel
		
		return _WrapingAsyncIterReceiveChannel(source1)
	
	# Handle synchronous iterables (and try to deduce the length in each case possible)
	if isinstance(channel, collections.abc.Iterable):
		count:  typing.Optional[int]  = None
		source2: typing.Iterable[T_co] = channel
		
		if isinstance(source2, collections.abc.Sequence):
			count = len(source2)
		
		return _WrapingSyncIterReceiveChannel(source2, count)
	
	assert False, "Unreachable code"



class ReceiveStream(trio.abc.ReceiveStream):
	"""A slightly extended version of `trio`'s standard interface for receiving byte streams.
	
	Attributes
	----------
	size
		The size of the entire stream data in bytes, or `None` if unavailable
	atime
		Time of the entry's last access (before the current one) in seconds
		since the Unix epoch, or `None` if unkown
	mtime
		Time of the entry's last modification in seconds since the Unix epoch,
		or `None` if unknown
	btime
		Time of entry creation in seconds since the Unix epoch, or `None`
		if unknown
	"""
	__slots__ = ("size", "atime", "mtime", "btime")
	
	# The total length of this stream (if available)
	size: typing.Optional[int]
	
	# The backing record's last access time
	atime: typing.Optional[typing.Union[int, float]]
	# The backing record's last modification time
	mtime: typing.Optional[typing.Union[int, float]]
	# The backing record's creation (“birth”) time
	btime: typing.Optional[typing.Union[int, float]]
	
	
	def __init__(self):
		self.size  = None
		self.atime = None
		self.mtime = None
		self.btime = None
	
	
	async def collect(self) -> bytes:
		value = bytearray()
		async with self:
			# Use “size”, if available, to try and read the entire stream's conents
			# in one go
			max_bytes = getattr(self, "size", None)
			
			while True:
				chunk = await self.receive_some(max_bytes)
				if len(chunk) < 1:
					break
				value += chunk
		return bytes(value)


class TeeingReceiveStream(ReceiveStream):
	"""Allows the value retrieved from a single `trio.abc.ReceiveStream` to be
	pushed towards several receivers additionally to be received by the caller
	it is returned to.
	"""
	
	__slots__ = ("_nursery", "_nursery_manager", "_channels", "_bufsize", "_closed", "_source")
	
	_nursery:         trio._core._run.Nursery  # No public type for this
	_nursery_manager: trio._core._run.NurseryManager  # No public type for this
	
	_channels: typing.List[trio.abc.SendChannel[bytes]]
	
	_bufsize: int
	_closed:  bool
	_source:  typing.Optional[trio.abc.ReceiveStream]
	
	def __init__(self, source: trio.abc.ReceiveStream, buffer_size: int = 0):
		super().__init__()
		
		self._bufsize = buffer_size
		self._source  = source
		self._closed  = False
		
		# Try to copy extra attributes from source stream
		self.size  = getattr(source, "size", None)
		self.atime = getattr(source, "atime", None)
		self.mtime = getattr(source, "mtime", None)
		self.btime = getattr(source, "btime", None)
		
		# Create nursery without using a `async with`-statement
		# (Only works because the `__aenter__`-call does not actually block on anything.)
		self._channels = []
		self._nursery_manager = trio.open_nursery()
		try:
			self._nursery_manager.__aenter__().__await__().send(None)
		except StopIteration as exc:
			self._nursery = exc.args[0]
		else:
			raise RuntimeError("Failed to initialize nursery synchronously")
	
	
	async def start_task(self, func: typing.Callable[[trio.abc.ReceiveStream], T], *args) -> T:
		send_channel, receive_channel = trio.open_memory_channel(self._bufsize)
		result = await self._nursery.start(func, receive_stream_from(receive_channel), *args)
		self._channels.append(send_channel)
		return result
	
	
	def start_task_soon(
			self, func: typing.Callable[[trio.abc.ReceiveStream], typing.Any], *args
	) -> None:
		send_channel, receive_channel = trio.open_memory_channel(self._bufsize)
		self._nursery.start_soon(func, receive_stream_from(receive_channel), *args)
		self._channels.append(send_channel)
	
	
	async def receive_some(self, max_bytes: typing.Optional[int] = None) -> bytes:
		if self._closed:
			raise trio.ClosedResourceError()
		if self._source is None:
			return b""
		
		try:
			value = await self._source.receive_some(max_bytes)
			
			# Pass received value (or EOF) to waiting write tasks
			if len(value) > 0:
				for channel in self._channels:
					await channel.send(value)
			else:
				for channel in self._channels:
					await channel.aclose()
				self._channels.clear()

				# Ensure that our slaves have finished before the final value
				# is returned
				await self.aclose(_mark_closed=False)
			
			return value
		except BaseException:
			await self.aclose()
			raise
	
	
	async def aclose(self, *, _mark_closed=True) -> None:
		if _mark_closed:
			self._closed = True
		if self._source is None:
			return
		
		try:
			try:
				# Close all remaining teeing streams by sending them a
				# cancellation error
				for channel in self._channels:
					with trio.CancelScope() as cancel_scope:
						cancel_scope.shield = True
						await trio.aclose_forcefully(channel)
				
				# Close the source stream
				await self._source.aclose()
			except BaseException as exc:
				# Wait for any tasks possibly still active in the nursery
				#
				# This must be called in this special `try` way to ensure that
				# the nursery will be properly closed down even if its contents
				# cannot be cleaned up anymore. Additionally, this will replace
				# a `trio.Cancelled` resulting of some other temporarily stored
				# exception by the actual exception value originally raised.
				if not await self._nursery_manager.__aexit__(type(exc), exc, exc.__traceback__):
					raise
			else:
				await self._nursery_manager.__aexit__(None, None, None)
		finally:
			self._channels.clear()
			self._source = None


class _WrapingTrioReceiveStream(ReceiveStream):
	"""
	Abstracts over a bare `trio.abc.ReceiveStream` to add our standard fields
	and methods to that stream
	"""
	__slots__ = ("_source",)
	
	_source: trio.abc.ReceiveStream
	
	def __init__(self, source: trio.abc.ReceiveStream):
		super().__init__()
		
		self._source = source
	
	
	async def receive_some(self, max_bytes: typing.Optional[int] = None) -> bytes:
		return await self._source.receive_some(max_bytes)
	
	
	async def aclose(self) -> None:
		await self._source.aclose()



class _WrapingIterReceiveStreamBase(ReceiveStream, typing.Generic[T_co]):
	"""Abstracts over various forms of synchronous and asynchronous returning of
	   byte streams
	"""
	
	__slots__ = ("_buffer", "_memview", "_offset", "_closed", "_source")
	
	_buffer:  bytearray
	_memview: typing.Optional[memoryview]
	_offset:  int
	_closed:  bool
	
	_source: typing.Optional[T_co]
	
	def __init__(self, source: T_co):
		super().__init__()
		
		self._source = source
		
		self._buffer  = bytearray()
		self._memview = None
		self._offset  = 0
		self._closed  = False
	
	
	@abc.abstractmethod
	async def _receive(self, max_bytes: typing.Optional[int]) -> bytes:
		pass
	
	
	async def receive_some(self, max_bytes: typing.Optional[int] = None) -> bytes:
		if self._closed:
			raise trio.ClosedResourceError()
		if self._source is None:
			return b""
		
		# Serve chunks from buffer if there is any data that hasn't been
		# delivered yet
		if self._memview:
			if max_bytes is not None:
				end_offset = min(self._offset + max_bytes, len(self._memview))
			else:
				end_offset = len(self._memview)
			
			result = bytes(self._memview[self._offset:end_offset])
			if end_offset >= len(self._memview):
				self._offset = 0
				self._memview.release()  # type: ignore  # Fixed in mypy 0.350
				self._memview = None
				self._buffer.clear()
			return result
		
		
		value = await self._receive(max_bytes)
		
		assert isinstance(value, bytes), \
		       f"Source stream {repr(self._source)} returned non-byte segment"
		
		if len(value) < 1:
			# We're at the end
			await self.aclose(_mark_closed=False)
			return b""
		
		# Stash extra bytes that are too large for our receiver
		if max_bytes is not None and max_bytes > len(value):
			self._buffer += value[max_bytes:]
			self._memview = memoryview(self._buffer)
			value = value[:max_bytes]
		
		return value
	
	
	@abc.abstractmethod
	async def _close_source(self) -> None:
		pass
	
	
	async def aclose(self, *, _mark_closed=True) -> None:
		if not self._closed and _mark_closed:
			self._closed = True
		
		if self._source is None:
			return
		
		try:
			await self._close_source()
		finally:
			self._source = None
			if self._memview is not None:
				self._memview.release()  # type: ignore  # Fixed in mypy 0.450
				self._memview = None
			self._buffer.clear()


class _WrapingAsyncIterReceiveStream(_WrapingIterReceiveStreamBase[typing.AsyncIterator[bytes]]):
	def __init__(self, source: typing.AsyncIterable[bytes]):
		super().__init__(source.__aiter__())
	
	
	async def _receive(self, _: typing.Optional[int]) -> bytes:
		assert self._source is not None
		
		# Skip empty returned byte strings as they have a special meaning here
		value = b""
		while len(value) < 1:
			try:
				value = await self._source.__anext__()
			except StopAsyncIteration:
				return b""
		return value
	
	
	async def _close_source(self) -> None:
		try:
			await self._source.aclose()  # type: ignore  # We catch errors instead
		except AttributeError:
			pass



class _WrapingSyncIterReceiveStream(_WrapingIterReceiveStreamBase[typing.Iterator[bytes]]):
	def __init__(self, source: typing.Iterable[bytes], size_hint: typing.Optional[int]):
		super().__init__(iter(source))
		
		self.size = size_hint
	
	
	async def _receive(self, _: typing.Optional[int]) -> bytes:
		assert self._source is not None
		
		# Skip empty returned byte strings as they have a special meaning here
		value = b""
		while len(value) < 1:
			try:
				value = next(self._source)
			except StopIteration:
				return b""
		return value
	
	
	async def _close_source(self) -> None:
		try:
			# We catch errors instead
			self._source.close()  # type: ignore[union-attr] # noqa: F821
		except AttributeError:
			pass



def receive_stream_from(stream: ArbitraryReceiveStream) -> ReceiveStream:
	# Optimization: Reuse given stream object, rather then creating a new
	#               wrapper when it is already of the right interface type
	if isinstance(stream, ReceiveStream):
		return stream
	
	# Optimization: Wrap the given Trio stream object in a tiny wrapper that
	#               just passes through all calls, but adds our extra fields
	#               and values
	if isinstance(stream, trio.abc.ReceiveStream):
		return _WrapingTrioReceiveStream(stream)
	
	# Handle asynchronous iterables
	if isinstance(stream, (collections.abc.AsyncIterable, collections.abc.Awaitable)):
		source1: typing.AsyncIterable[bytes]
		
		# Wrap awaitables of bytes in an asynchronous iterable that yields once
		if isinstance(stream, collections.abc.Awaitable):
			async def await_iter_wrapper(stream: typing.Awaitable[bytes]) \
			      -> typing.AsyncIterable[bytes]:
				yield await stream
			source1 = await_iter_wrapper(stream)
		else:
			source1 = stream
		
		return _WrapingAsyncIterReceiveStream(source1)
	
	# Handle synchronous iterables (and try to deduce the length in each case possible)
	if isinstance(stream, (bytes, collections.abc.Iterable)):
		size:    typing.Optional[int] = None
		source2: typing.Iterable[bytes]
		
		# Wrap a simple bytes value in a tuple to make it an iterable
		if isinstance(stream, bytes):
			source2 = (stream,)
		else:
			source2 = stream
		
		# Deduce the length of sequences of bytes (including the single bytes sequence above)
		if isinstance(source2, collections.abc.Sequence):
			size = sum(len(item) for item in source2)
		# … and in-memory byte stream files as well
		elif isinstance(source2, io.BytesIO):
			# Asks for just the remaining length, restoring the position afterwards
			pos = source2.tell()
			source2.seek(0, io.SEEK_END)
			size = source2.tell() - pos
			source2.seek(pos, io.SEEK_SET)
		
		return _WrapingSyncIterReceiveStream(source2, size)
	
	assert False, "Unreachable code"

A slightly extended version of `trio`'s standard interface for receiving object streams. 	 	Attributes 	---------- 	count 		The number of objects that will be returned, or `None` if unavailable 	atime 		Time of the entry's last access (before the current one) in seconds 		since the Unix epoch, or `None` if unkown 	mtime 		Time of the entry's last modification in seconds since the Unix epoch, 		or `None` if unknown 	btime 		Time of entry creation in seconds since the Unix epoch, or `None` 		if unknown # The total length of this stream (if available) # The backing record's last access time # The backing record's last modification time # The backing record's creation (“birth”) time # No public type for this # No public type for this Allows the value retrieved from a single `trio.abc.ReceiveChannel` to be 	pushed towards several receivers additionally to be received by the caller 	it is returned to. # Try to copy extra attributes from source channel # Create nursery without using a `async with`-statement # (Only works because the `__aenter__`-call does not actually block on anything.) # type: ignore[attr-defined] # upstream type bug # noqa: F821 # Doing this sync is just wrong, but we cannot block in this function… # Cast source value to ignore the possible `None` variant as the # passed source value will be ignored if we provide `_shared` # Pass received value (or EOF) to waiting write tasks # type: ignore[attr-defined] # upstream type bug # noqa: F821 # Ensure that our slaves have finished before the final value is # returned # We implement this as there is no guarantee that *all* of our teeing # channels will accept the received value non-blockingly # Close all remaining teeing streams by sending them a # cancellation error # Close the source stream # Wait for any tasks possibly still active in the nursery # # This must be called in this special `try` way to ensure that # the nursery will be properly closed down even if its contents # cannot be cleaned up anymore. Additionally, this will replace # a `trio.Cancelled` resulting of some other temporarily stored # exception by the actual exception value originally raised. Abstracts over various forms of synchronous and asynchronous returning of 	   object streams # type: ignore[attr-defined]  # upstream type bug # noqa: F821 # Cannot ask this stream type for a non-blocking value # type: ignore  # We catch errors instead # We cannot handle invoking async close here # type: ignore  # We catch errors instead # Optimization: Reuse given stream object, rather then creating a new #               wrapper when it is already of the right interface type # Optimization: Wrap the given Trio stream object in a tiny wrapper that #               just passes through all calls, but adds our extra fields #               and values # Handle asynchronous iterables # Handle synchronous iterables (and try to deduce the length in each case possible) A slightly extended version of `trio`'s standard interface for receiving byte streams. 	 	Attributes 	---------- 	size 		The size of the entire stream data in bytes, or `None` if unavailable 	atime 		Time of the entry's last access (before the current one) in seconds 		since the Unix epoch, or `None` if unkown 	mtime 		Time of the entry's last modification in seconds since the Unix epoch, 		or `None` if unknown 	btime 		Time of entry creation in seconds since the Unix epoch, or `None` 		if unknown # The total length of this stream (if available) # The backing record's last access time # The backing record's last modification time # The backing record's creation (“birth”) time # Use “size”, if available, to try and read the entire stream's conents # in one go Allows the value retrieved from a single `trio.abc.ReceiveStream` to be 	pushed towards several receivers additionally to be received by the caller 	it is returned to. # No public type for this # No public type for this # Try to copy extra attributes from source stream # Create nursery without using a `async with`-statement # (Only works because the `__aenter__`-call does not actually block on anything.) # Pass received value (or EOF) to waiting write tasks # Ensure that our slaves have finished before the final value # is returned # Close all remaining teeing streams by sending them a # cancellation error # Close the source stream # Wait for any tasks possibly still active in the nursery # # This must be called in this special `try` way to ensure that # the nursery will be properly closed down even if its contents # cannot be cleaned up anymore. Additionally, this will replace # a `trio.Cancelled` resulting of some other temporarily stored # exception by the actual exception value originally raised. Abstracts over a bare `trio.abc.ReceiveStream` to add our standard fields 	and methods to that stream Abstracts over various forms of synchronous and asynchronous returning of 	   byte streams # Serve chunks from buffer if there is any data that hasn't been # delivered yet # type: ignore  # Fixed in mypy 0.350 # We're at the end # Stash extra bytes that are too large for our receiver # type: ignore  # Fixed in mypy 0.450 # Skip empty returned byte strings as they have a special meaning here # type: ignore  # We catch errors instead # Skip empty returned byte strings as they have a special meaning here # We catch errors instead # type: ignore[union-attr] # noqa: F821 # Optimization: Reuse given stream object, rather then creating a new #               wrapper when it is already of the right interface type # Optimization: Wrap the given Trio stream object in a tiny wrapper that #               just passes through all calls, but adds our extra fields #               and values # Handle asynchronous iterables # Wrap awaitables of bytes in an asynchronous iterable that yields once # Handle synchronous iterables (and try to deduce the length in each case possible) # Wrap a simple bytes value in a tuple to make it an iterable # Deduce the length of sequences of bytes (including the single bytes sequence above) # … and in-memory byte stream files as well # Asks for just the remaining length, restoring the position afterwards