Repository: jina-ai/serve
Pull request: feat: add needs_all()
Address https://github.com/jina-ai/jina/issues/1262
1. Add needs_all() to collect all hanging pods so far as a sugary syntax for .needs(['h1', 'h2', ...]), .needs_all() can be used at any part of the Flow.
2. add test_flow_needs_all() in unit test test_flow.py 

Problem statement:
add a sugary Flow API: needs_all()
**Describe the feature**

Expected behaviors:
- `Flow()....needs_all()` collect all hanging Pod so far, it is a sugary syntax for `.needs(['h1', 'h2', ...])`
- Like `.needs()`, `.needs_all()` can be used at any part of the Flow
- All other behaviors follow `needs()`

Hanging Pods are the pods that have no output **so far**.

**Your proposal**
<!-- copy past your code/pull request link -->

---
<!-- Optional, but really help us locate the problem faster -->

**Environment**
<!-- Run `jina --version-full` and copy paste the output here -->

**Screenshots**
<!-- If applicable, add screenshots to help explain your problem. -->

Hints from issue discussion:

Commit message: test: add needs_all() unit test

Patch to review:
diff --git a/jina/flow/__init__.py b/jina/flow/__init__.py
index f55d34af57807..0cfa6264befae 100644
--- a/jina/flow/__init__.py
+++ b/jina/flow/__init__.py
@@ -251,6 +251,19 @@ def needs(self, needs: Union[Tuple[str], List[str]],
             raise FlowTopologyError('no need to wait for a single service, need len(needs) > 1')
         return self.add(name=name, needs=needs, pod_role=PodRoleType.JOIN, *args, **kwargs)
 
+    def needs_all(self, copy_flow: bool = True,
+                  name: str = 'joiner', *args, **kwargs) -> 'Flow':
+        """
+        Collect all hanging Pod so far and add a blocker to the flow, wait until all handing peas completed.
+        :param copy_flow: when set to true, then always copy the current flow and do the modification on top of it then return, otherwise, do in-line modification
+        :param name: the name of this joiner, by default is ``joiner``
+        :return: the modified flow
+        """
+        op_flow = copy.deepcopy(self) if copy_flow else self
+        needs = _hanging_pods(op_flow)
+
+        return self.needs(needs=needs, name=name, *args, **kwargs)
+
     def add(self,
             needs: Union[str, Tuple[str], List[str]] = None,
             copy_flow: bool = True,
diff --git a/tests/unit/flow/test_flow.py b/tests/unit/flow/test_flow.py
index e0ff4ef872bae..d361d108f811b 100644
--- a/tests/unit/flow/test_flow.py
+++ b/tests/unit/flow/test_flow.py
@@ -559,3 +559,17 @@ def test_flow_arbitrary_needs():
 
     with f:
         f.index_lines(['abc', 'def'])
+
+
+def test_flow_needs_all():
+    f = (Flow().add(name='p1', needs='gateway')
+         .add(name='p2', needs='gateway')
+         .add(name='p3', needs='gateway')
+         .needs(needs=['p1', 'p2'], name='r1')
+         .needs_all(name='r2')
+         )
+
+    all_needs = {v for p in f._pod_nodes.values() for v in p.needs}
+    all_names = {p for p in f._pod_nodes.keys()}
+
+    assert list(all_names.difference(all_needs)) == ['r2']
\ No newline at end of file


Code review comments:
diff_hunk: @@ -251,6 +251,19 @@ def needs(self, needs: Union[Tuple[str], List[str]],
             raise FlowTopologyError('no need to wait for a single service, need len(needs) > 1')
         return self.add(name=name, needs=needs, pod_role=PodRoleType.JOIN, *args, **kwargs)
 
+    def needs_all(self, copy_flow: bool = True,
+                  name: str = 'joiner', *args, **kwargs) -> 'Flow':
+        """
+        Collect all hanging Pod so far and add a blocker to the flow, wait until all handing peas completed.
+        :param copy_flow: when set to true, then always copy the current flow and do the modification on top of it then return, otherwise, do in-line modification
+        :param name: the name of this joiner, by default is ``joiner``
+        :return: the modified flow
+        """
+        op_flow = copy.deepcopy(self) if copy_flow else self
+        needs = _hanging_pods(op_flow)
+
+        return self.needs(needs=needs, name=name, *args, **kwargs)
path: jina/flow/__init__.py
text: @user1:
This won't work, if there is only one hanging pod. Thus, you can either add a flag `accept_one_pod` to the `needs` function in order to reuse `needs` or directly call `add` here.

@author:
I have updated the code to fix these issues accordingly and also added required unit tests, please let me know if that works. Thanks!

@user1:
Looks great so far. One tiny cleanup: Could you switch the order of variables in the calling of `add` and `needs` to be the same? That'll improve readability.

@author:
Sure! Thank you for the advice! 
I have made the change to rearrange them and keep it consistent. 
I have also thought a little about it before and the reason I arranged the order of variables this way is that I wanted to keep it consistent with current conventions. (e.g. In the unit test test_flow_arbitrary_needs(), .add() is called as (name, needs) and .needs is called as (needs, name).

diff_hunk: @@ -251,6 +251,19 @@ def needs(self, needs: Union[Tuple[str], List[str]],
             raise FlowTopologyError('no need to wait for a single service, need len(needs) > 1')
         return self.add(name=name, needs=needs, pod_role=PodRoleType.JOIN, *args, **kwargs)
 
+    def needs_all(self, copy_flow: bool = True,
+                  name: str = 'joiner', *args, **kwargs) -> 'Flow':
+        """
+        Collect all hanging Pod so far and add a blocker to the flow, wait until all handing peas completed.
+        :param copy_flow: when set to true, then always copy the current flow and do the modification on top of it then return, otherwise, do in-line modification
+        :param name: the name of this joiner, by default is ``joiner``
+        :return: the modified flow
+        """
+        op_flow = copy.deepcopy(self) if copy_flow else self
path: jina/flow/__init__.py
text: @user1:
There is no need to `deepcopy` the flow here. `_hanging_pods` will not change the flow in any way.