# Path: azmq/multiplexer.py
# class Multiplexer(ClosableAsyncObject):
#     def on_open(self):
#         super().on_open()
#         self._sockets = set()
# 
#     def add_socket(self, socket):
#         """
#         Add a socket to the multiplexer.
# 
#         :param socket: The socket. If it was added already, it won't be added a
#             second time.
#         """
#         if socket not in self._sockets:
#             self._sockets.add(socket)
#             socket.on_closed.connect(self.remove_socket)
# 
#     def remove_socket(self, socket):
#         """
#         Remove a socket from the multiplexer.
# 
#         :param socket: The socket. If it was removed already or if it wasn't
#             added, the call does nothing.
#         """
#         if socket in self._sockets:
#             socket.on_closed.disconnect(self.remove_socket)
#             self._sockets.remove(socket)
# 
#     @cancel_on_closing
#     async def recv_multipart(self):
#         """
#         Read from all the associated sockets.
# 
#         :returns: A list of tuples (socket, frames) for each socket that
#             returned a result.
#         """
#         if not self._sockets:
#             return []
# 
#         results = []
# 
#         async def recv_and_store(socket):
#             frames = await socket.recv_multipart()
#             results.append((socket, frames))
# 
#         tasks = [
#             asyncio.ensure_future(recv_and_store(socket), loop=self.loop)
#             for socket in self._sockets
#         ]
# 
#         try:
#             await asyncio.wait(
#                 tasks,
#                 return_when=asyncio.FIRST_COMPLETED,
#                 loop=self.loop,
#             )
#         finally:
#             for task in tasks:
#                 task.cancel()
# 
#         return results
# 
# Path: azmq/errors.py
# class InvalidOperation(AZMQError):
#     def __init__(self, msg):
#         super().__init__(msg)
# 
# class ProtocolError(AZMQError):
#     def __init__(self, msg, fatal=False):
#         super().__init__(msg)
#         self.fatal = fatal
# 
# class UnsupportedSchemeError(AZMQError):
#     def __init__(self, scheme):
#         super().__init__("Unsupported scheme '%s'" % scheme)
#         self.scheme = scheme
# 
# Path: tests/conftest.py
# def event_loop():
# def reader():
#     async def readexactly(count):
# def writer():
# ENDPOINTS = [
#     'inproc://mypath',
#     'tcp://127.0.0.1:3333',
# ]


import asyncio
import pytest
import azmq
from azmq.multiplexer import Multiplexer
from azmq.errors import (
    InvalidOperation,
    ProtocolError,
    UnsupportedSchemeError,
)
from ..conftest import use_all_transports
        try:
            future = asyncio.Future(loop=event_loop)
            req_socket.on_connection_failure.connect(future.set_result)
            req_socket.connect(endpoint)
            rep_socket.identity = b'\0invalid'
            rep_socket.bind(endpoint)

            await fivesec(future)
            assert isinstance(future.result().error, ProtocolError)
            assert future.result().error.fatal

        finally:
            req_socket.close()
            rep_socket.close()


@use_all_transports
@pytest.mark.asyncio
async def test_req_req_invalid_combination(event_loop, endpoint):
    async with azmq.Context() as context:
        req_socket = context.socket(azmq.REQ)
        req_socket_2 = context.socket(azmq.REQ)

        try:
            future = asyncio.Future(loop=event_loop)
            req_socket.on_connection_failure.connect(future.set_result)
            req_socket.connect(endpoint)
            req_socket_2.bind(endpoint)

            await fivesec(future)

# Path: azmq/multiplexer.py
# class Multiplexer(ClosableAsyncObject):
#     def on_open(self):
#         super().on_open()
#         self._sockets = set()
# 
#     def add_socket(self, socket):
#         """
#         Add a socket to the multiplexer.
# 
#         :param socket: The socket. If it was added already, it won't be added a
#             second time.
#         """
#         if socket not in self._sockets:
#             self._sockets.add(socket)
#             socket.on_closed.connect(self.remove_socket)
# 
#     def remove_socket(self, socket):
#         """
#         Remove a socket from the multiplexer.
# 
#         :param socket: The socket. If it was removed already or if it wasn't
#             added, the call does nothing.
#         """
#         if socket in self._sockets:
#             socket.on_closed.disconnect(self.remove_socket)
#             self._sockets.remove(socket)
# 
#     @cancel_on_closing
#     async def recv_multipart(self):
#         """
#         Read from all the associated sockets.
# 
#         :returns: A list of tuples (socket, frames) for each socket that
#             returned a result.
#         """
#         if not self._sockets:
#             return []
# 
#         results = []
# 
#         async def recv_and_store(socket):
#             frames = await socket.recv_multipart()
#             results.append((socket, frames))
# 
#         tasks = [
#             asyncio.ensure_future(recv_and_store(socket), loop=self.loop)
#             for socket in self._sockets
#         ]
# 
#         try:
#             await asyncio.wait(
#                 tasks,
#                 return_when=asyncio.FIRST_COMPLETED,
#                 loop=self.loop,
#             )
#         finally:
#             for task in tasks:
#                 task.cancel()
# 
#         return results
# 
# Path: azmq/errors.py
# class InvalidOperation(AZMQError):
#     def __init__(self, msg):
#         super().__init__(msg)
# 
# class ProtocolError(AZMQError):
#     def __init__(self, msg, fatal=False):
#         super().__init__(msg)
#         self.fatal = fatal
# 
# class UnsupportedSchemeError(AZMQError):
#     def __init__(self, scheme):
#         super().__init__("Unsupported scheme '%s'" % scheme)
#         self.scheme = scheme
# 
# Path: tests/conftest.py
# def event_loop():
# def reader():
#     async def readexactly(count):
# def writer():
# ENDPOINTS = [
#     'inproc://mypath',
#     'tcp://127.0.0.1:3333',
# ]


# Path: tests/integration/test_scenarios.py
import asyncio
import pytest
import azmq
from azmq.multiplexer import Multiplexer
from azmq.errors import (
    InvalidOperation,
    ProtocolError,
    UnsupportedSchemeError,
)
from ..conftest import use_all_transports

        try:
            future = asyncio.Future(loop=event_loop)
            req_socket.on_connection_failure.connect(future.set_result)
            req_socket.connect(endpoint)
            rep_socket.identity = b'\0invalid'
            rep_socket.bind(endpoint)

            await fivesec(future)
            assert isinstance(future.result().error, ProtocolError)
            assert future.result().error.fatal

        finally:
            req_socket.close()
            rep_socket.close()


@use_all_transports
@pytest.mark.asyncio
async def test_req_req_invalid_combination(event_loop, endpoint):
    async with azmq.Context() as context:
        req_socket = context.socket(azmq.REQ)
        req_socket_2 = context.socket(azmq.REQ)

        try:
            future = asyncio.Future(loop=event_loop)
            req_socket.on_connection_failure.connect(future.set_result)
            req_socket.connect(endpoint)
            req_socket_2.bind(endpoint)

            await fivesec(future)

            assert isinstance(future.result().error, ProtocolError)