Header("content-type", "text/plain"),
      )
    stream.writeHeaders(
      responseHeaders = responseHeaders,
      outFinished = false,
      flushHeaders = false,
    )
    val out = stream.getSink().buffer()
    out.writeUtf8("Not found: $path")
    out.close()
  }

  private fun serveDirectory(
    stream: Http2Stream,
    files: Array<File>,
  ) {
    val responseHeaders =
      listOf(

internal class RealStream(
  private val http2Stream: Http2Stream,
) : Stream {
  override val requestBody = http2Stream.getSource().buffer()
  override val responseBody = http2Stream.getSink().buffer()

  override fun cancel() {
    http2Stream.closeLater(ErrorCode.CANCEL)
  }

    val stream2 = connection.newStream(headerEntries("b", "banana"), true)
    connection.writePingAndAwaitPong() // Ensure the GO_AWAY that resets stream2 has been received.
    val sink1 = stream1.getSink().buffer()
    val sink2 = stream2.getSink().buffer()
    sink1.writeUtf8("abc")
    assertFailsWith<IOException> {
      sink2.writeUtf8("abc")
      sink2.flush()
    }.also { expected ->

    } else {
      Protocol.HTTP_2
    }

  @Volatile
  private var canceled = false

  override fun createRequestBody(
    request: Request,
    contentLength: Long,
  ): Sink {
    return stream!!.getSink()
  }

  override fun writeRequestHeaders(request: Request) {
    if (stream != null) return

    val hasRequestBody = request.body != null
    val requestHeaders = http2HeadersList(request)

	zlog := &zapLogger{
		l: zl.l,
	}

	return logr.New(zlog)
}

func (zl *zapLogger) WithValues(keysAndValues ...any) logr.LogSink {
	return NewLogrAdapter(zl.l.WithLabels(keysAndValues...)).GetSink()
}

func (zl *zapLogger) WithName(string) logr.LogSink {
	return zl
}

// NewLogrAdapter creates a new logr.Logger using the given Zap Logger to log.
func NewLogrAdapter(l *Scope) logr.Logger {

  // If op is a `tf_executor.NextIteration.Source`, visit its associated
  // `tf_executor.NextIteration.Sink` op.
  if (auto source_op = llvm::dyn_cast<NextIterationSourceOp>(op)) {
    Operation* sink_op = source_op.GetSink().getOperation();
    if (reachable_ops->insert(sink_op).second) ops_to_visit->push_back(sink_op);
  }
}

GraphPruningPass::GraphPruningPass(
    llvm::ArrayRef<std::string> ops_to_preserve) {

  // pair has matching devices or no devices.
  if (auto source = llvm::dyn_cast<tf_executor::NextIterationSourceOp>(op)) {
    return ops_have_same_device(source, source.GetSink());
  } else if (auto sink = llvm::dyn_cast<tf_executor::NextIterationSinkOp>(op)) {
    return ops_have_same_device(sink.GetSource(), sink);
  }

  return llvm::isa<tf_executor::EnterOp, tf_executor::ExitOp,

				t.Errorf("managerImpl.processShutdownEvent() error = %v, wantErr %v", err, tt.wantErr)
			}

			underlier, ok := logger.GetSink().(ktesting.Underlier)
			if !ok {
				t.Fatalf("Should have had a ktesting LogSink, got %T", logger.GetSink())
			}

			log := underlier.GetBuffer().String()
			if !strings.Contains(log, tt.expectedOutputContains) {
				// Log will be shown on failure. To see it

      $_state.types = { result_type, token_type, control_type };
      $_state.attributes.append(attributes.begin(), attributes.end());
    }]>
   ];

  let extraClassDeclaration = [{
    NextIterationSinkOp GetSink() {
      return cast<NextIterationSinkOp>(*getToken().user_begin());
    }
  }];

  let assemblyFormat = "`:` type($output) attr-dict";
  let hasCustomAssemblyFormat = 0;
}

  /**
   * Returns a sink that can be used to write data to the peer.
   *
   * @throws IllegalStateException if this stream was initiated by the peer and a [writeHeaders] has
   *     not yet been sent.
   */
  fun getSink(): Sink {
    this.withLock {
      check(hasResponseHeaders || isLocallyInitiated) {
        "reply before requesting the sink"
      }
    }
    return sink
  }

  /**