feat: Support communicating with individual Executors in CustomGateway

[Jackmin801]

Goals:

• resolves Support communicating with individual Executors with GatewayStreamer #5538
• Add tests
• Implement intended API
• check and update documentation. See guide and ask the team.

[Jackmin801]

Gateway code:

from docarray import DocumentArray, Document

from jina.serve.runtimes.gateway.http.fastapi import FastAPIBaseGateway
from jina import Flow, Executor, requests

class MyGateway(FastAPIBaseGateway):
    @property
    def app(self):
        from fastapi import FastAPI

        app = FastAPI(title='Custom FastAPI Gateway')

        @app.get("/endpoint")
        async def get(text: str):
            docs = await self.executor['executor1'](exec_endpoint='/', docs=DocumentArray([Document(text=text)]), parameters={'k': 'v'})
            return {'result': docs[0].text}

        return app


class FirstExec(Executor):
    @requests
    def func(self, docs, **kwargs):
        for doc in docs:
            doc.text += " First"

class SecondExec(Executor):
    @requests
    def func(self, docs, **kwargs):
        print(kwargs)
        for doc in docs:
            doc.text += " Second"

with Flow(port=12345).config_gateway(uses=MyGateway, protocol='http', port=50975).add(
    uses=FirstExec, name='executor0'
).add(
    uses=SecondExec, name='executor1'
) as flow:
    flow.block()

Client code

curl 'localhost:50975/endpoint?text=meow'

[codecov]

Codecov Report

Merging #5558 (73a94b6) into master (a5ef69b) will decrease coverage by 15.58%.
The diff coverage is 33.33%.

@@             Coverage Diff             @@
##           master    #5558       +/-   ##
===========================================
- Coverage   86.17%   70.59%   -15.59%     
===========================================
  Files         124      122        -2     
  Lines       10048    10022       -26     
===========================================
- Hits         8659     7075     -1584     
- Misses       1389     2947     +1558     

Flag	Coverage Δ
jina	70.59% <33.33%> (-15.59%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
jina/serve/streamer.py	77.65% <26.31%> (-13.01%)	⬇️
jina/serve/gateway.py	92.18% <100.00%> (-3.06%)	⬇️
jina/schemas/flow.py	0.00% <0.00%> (-100.00%)	⬇️
jina/schemas/meta.py	0.00% <0.00%> (-100.00%)	⬇️
jina/schemas/gateway.py	0.00% <0.00%> (-100.00%)	⬇️
jina/schemas/__init__.py	0.00% <0.00%> (-100.00%)	⬇️
jina/schemas/executor.py	0.00% <0.00%> (-100.00%)	⬇️
jina/schemas/deployment.py	0.00% <0.00%> (-100.00%)	⬇️
jina/schemas/helper.py	0.00% <0.00%> (-91.67%)	⬇️
jina/checker.py	0.00% <0.00%> (-85.72%)	⬇️
... and 69 more

Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here.

[alaeddine-13]

I suggest 2 options:

• keep introducing ExecutorStreamer, I think in this case it makes sense to rename GatewayStreamer to FlowStreamer. Since we are making a breaking change, it makes sense to unify the interface as well. This means having:

• self.flow_streamer.post(...) # or self.flow.post(...)
• self.executor_streamer[exec].post(...) # or self.executors[exec].post(...)

• unify flow streamer and executor streamer under the same self.streamer object. I think this makes sense because otherwise, since the same pool is used, destroying one object corrupts the other

[Jackmin801]

zz bot

[JoanFM]

no, I do not think so

[JoanFM]

Technically approve, wait for @alexcg1 for the docs

[alexcg1]

LGTM 👍

[github-actions]

📝 Docs are deployed on https://feat-5538-comm-with-individual-execs--jina-docs.netlify.app 🎉