code: @Override
    public <T> void stream(HeadersAmp headers, 
                           ResultStream<T> result,
                           StubAmp actor,
                           Object[] args)
    {
      int nodeCount = _serviceRef.getPod().nodeCount();
      
      if (nodeCount <= 1) {
        try {
          MethodShim methodShim = findActiveMethodShim();
          
          methodShim.stream(headers, result, actor, args);
        } catch (Throwable exn) {
          result.fail(exn);
        }
        return;
      }
      
      //ResultStreamJoin<T> joinLocal = null;
      
      ResultStreamAmp<T> resultAmp = (ResultStreamAmp<T>) result;
  
      // joinLocal = new ResultStreamJoin<>(result, _serviceRef.getSelfInbox());
      
      //ResultStream<T> createFork = joinLocal.fork();
      //ResultStream<T> createFork = resultAmp.fork();
  
      for (int i = 0; i < nodeCount; i++) {
        MethodRefActive method = findMethodRefActive(i);
  
        if (method != null) {
          // ResultStream<T> resultFork = joinLocal.fork();
          ResultStream<T> resultFork = resultAmp.fork();
  
          // MethodRefAmp methodRef = method.getMethodRef();
          //ActorAmp actorMessage = method.getMethodRef().getActor(actor);
          //ActorAmp actorMessage = method.getMethodRef().getService().getActor();
          //actorMessage = method.getMethodRef().getActor(actorMessage);
  
          method.getMethodShim().stream(headers, resultFork, actor, args);
          //methodRef.stream(headers, resultFork, args);
        }
        else {
          RuntimeException exn = new IllegalStateException();
          result.fail(exn);
        }
      }
  
      resultAmp.forkComplete();
    }
docs: stream call with fork/join to all nodes in the pod.