NodeVisitState state = visitedNodes.get(node);
    if (state == NodeVisitState.COMPLETE) {
      return false;
    }
    if (state == NodeVisitState.PENDING) {
      return true;
    }

    visitedNodes.put(node, NodeVisitState.PENDING);
    for (N nextNode : graph.successors(node)) {
      if (canTraverseWithoutReusingEdge(graph, nextNode, previousNode)

		select {
		case <-m.ctx.Done():
			if debugPrint {
				fmt.Println("Client sending disconnect to mux", m.MuxID)
			}
			m.addErrorNonBlockingClose(internalResp, context.Cause(m.ctx))
			errState = true
			continue
		case req, ok := <-requests:
			if !ok {
				// Done send EOF
				if debugPrint {
					fmt.Println("Client done, sending EOF to mux", m.MuxID)
				}
				msg := message{
					Op:    OpMuxClientMsg,

                        master.put(key, element);

                        if (!pending.isEmpty()) {
                            predecessors.put(key, pending);
                            pending = new ArrayList<>();
                        }
                    } else {
                        pending.add(element);
                    }
                }

        "items:read items:write users:read profile openid"
    client_id: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
        using HTTP Basic auth, as: client_id:client_secret
    client_secret: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
        using HTTP Basic auth, as: client_id:client_secret
    """

    def __init__(

			fd := int(wl.Netns.Fd())
			log.Infof("Sending local pod %s ztunnel", uid)
			resp, err = conn.sendMsgAndWaitForAck(&zdsapi.WorkloadRequest{
				Payload: &zdsapi.WorkloadRequest_Add{
					Add: &zdsapi.AddWorkload{
						Uid:          uid,
						WorkloadInfo: wl.Workload,
					},
				},
			}, &fd)
		} else {
			log.Infof("netns not available for local pod %s. sending keep to ztunnel", uid)

  // refers to; it MUST be the same name returned by the CSI GetPluginName()
  // call for that driver.
  // The driver name must be 63 characters or less, beginning and ending with
  // an alphanumeric character ([a-z0-9A-Z]) with dashes (-), dots (.), and
  // alphanumerics between.
  // More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata

  {{.Name}} - {{.Usage}}

USAGE:
  {{.Name}} {{if .VisibleFlags}}[FLAGS]{{end}} METAFILES...

Multiple files can be added. Files ending in '.zip' will be searched
for 'xl.meta' files. Wildcards are accepted: 'testdir/*.txt' will compress
all files in testdir ending with '.txt', directories can be wildcards
as well. 'testdir/*/*.txt' will match 'testdir/subdir/b.txt', double stars

   * have been received. If the returned list contains multiple blocks of headers the blocks will be
   * delimited by 'null'.
   *
   * @param callerIsIdle true if the caller isn't sending any more bytes until the peer responds.
   *     This is true after a `Expect-Continue` request, false for duplex requests, and false for
   *     all other requests.
   */
  @Throws(IOException::class)

  // refers to; it MUST be the same name returned by the CSI GetPluginName()
  // call for that driver.
  // The driver name must be 63 characters or less, beginning and ending with
  // an alphanumeric character ([a-z0-9A-Z]) with dashes (-), dots (.), and
  // alphanumerics between.
  // More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata

			t.Errorf("want %q, got %q", testPayload+strconv.Itoa(n), string(resp.Msg))
		}
		if n == 10 {
			close(stream.Requests)
			continue
		}
		n++
		t.Log("sending new client request")
		stream.Requests <- []byte(strconv.Itoa(n))
	}
	t.Log("EOF. 10 Roundtrips:", time.Since(start))
}

func testStreamCancel(t *testing.T, local, remote *Manager) {