Add parallelize option to async_w3.eth.subscribe API

- Add test for eth.subscribe with all kwargs and make sure they
  are passed through when creating the subscription
- Clean up old ``_subscribe_with_args``

Author: fselmo

tests/core/subscriptions/test_subscription_manager.py

@@ -490,3 +490,42 @@ async def parallel_handler(_ctx) -> None:
 
     assert subscription_manager.total_handler_calls == 3
     assert len(manager_tasks) == 0  # all tasks cleaned up
+
+
+@pytest.mark.asyncio
+async def test_eth_subscribe_api_call_with_all_kwargs(subscription_manager):
+    async_w3 = subscription_manager._w3
+    provider = subscription_manager._w3.provider
+
+    label = "test_subscription"
+    test_ctx = "test context"
+
+    async def fast_parallel_handler(context) -> None:
+        assert asyncio.current_task() in subscription_manager._tasks
+        assert context.test_ctx == test_ctx
+        sub = subscription_manager.get_by_id(context.subscription.id)
+        assert sub.label == label
+
+        await context.subscription.unsubscribe()
+
+    sub_id = await async_w3.eth.subscribe(
+        "newHeads",
+        handler=fast_parallel_handler,
+        handler_context={"test_ctx": test_ctx},
+        label=label,
+        parallelize=True,
+    )
+    provider._request_processor.cache_request_information(
+        request_id=sub_id,
+        method="eth_subscribe",
+        params=[],
+        response_formatters=((), (), ()),
+    )
+    provider._request_processor._handler_subscription_queue.put_nowait(
+        create_subscription_message(sub_id)
+    )
+
+    await subscription_manager.handle_subscriptions()
+
+    assert subscription_manager.total_handler_calls == 1
+    assert len(async_w3.subscription_manager._tasks) == 0

web3/eth/async_eth.py

@@ -720,19 +720,6 @@ async def uninstall_filter(self, filter_id: HexStr) -> bool:
         mungers=[default_root_munger],
     )
 
-    _subscribe_with_args: Method[
-        Callable[
-            [
-                SubscriptionType,
-                Optional[Union[LogsSubscriptionArg, bool]],
-            ],
-            Awaitable[HexStr],
-        ]
-    ] = Method(
-        RPC.eth_subscribe,
-        mungers=[default_root_munger],
-    )
-
     async def subscribe(
         self,
         subscription_type: SubscriptionType,
@@ -745,6 +732,7 @@ async def subscribe(
         handler: Optional[EthSubscriptionHandler] = None,
         handler_context: Optional[Dict[str, Any]] = None,
         label: Optional[str] = None,
+        parallelize: Optional[bool] = None,
     ) -> HexStr:
         if not isinstance(self.w3.provider, PersistentConnectionProvider):
             raise MethodNotSupported(
@@ -757,6 +745,7 @@ async def subscribe(
             handler=handler,
             handler_context=handler_context or {},
             label=label,
+            parallelize=parallelize,
         )
         return await self.w3.subscription_manager.subscribe(sub)