uint64 size) override {
    OwnedTFStatus c_status(TF_NewStatus());
    SP_Stream stream_handle = static_cast<CStream*>(stream)->Handle();
    SP_DeviceMemoryBase device_mem = DeviceMemoryBaseToC(location);
    stream_executor_->mem_zero(&device_, stream_handle, &device_mem, size,
                               c_status.get());
    return StatusFromTF_Status(c_status.get());
  }

        given:
        def streamsHandler = Mock(StreamsHandler)
        def execHandle = handle().args(args(TestApp.class)).setDisplayName("foo proc").streamsHandler(streamsHandler).build()

        when:
        execHandle.start()
        def result = execHandle.waitForFinish()

        then:
        result.rethrowFailure()
        1 * streamsHandler.connectStreams(_ as Process, "foo proc", _ as Executor)

      }

    /** Sets the response body to the UTF-8 encoded bytes of [body]. */
    fun body(body: String): Builder = body(Buffer().writeUtf8(body))

    fun streamHandler(streamHandler: StreamHandler) =
      apply {
        this.streamHandler = streamHandler
      }

    /**
     * Sets the response body to [body], chunked every [maxChunkSize] bytes.
     */
    fun chunkedBody(
      body: Buffer,

        .streamHandler(body1)
        .build(),
    )
    val body =
      MockStreamHandler()
        .sendResponse("response body\n")
        .exhaustResponse()
        .receiveRequest("request body\n")
        .exhaustRequest()
    server.enqueue(
      MockResponse.Builder()
        .clearHeaders()
        .streamHandler(body)
        .build(),
    )
    val call =

	public final fun status (Ljava/lang/String;)Lmockwebserver3/MockResponse$Builder;
	public final fun streamHandler (Lmockwebserver3/StreamHandler;)Lmockwebserver3/MockResponse$Builder;
	public final fun throttleBody (JJLjava/util/concurrent/TimeUnit;)Lmockwebserver3/MockResponse$Builder;
	public final fun trailers (Lokhttp3/Headers;)Lmockwebserver3/MockResponse$Builder;

    /**
     * The variables to set in the environment the executable is run in.
     */
    private final Map<String, String> environment;
    private final StreamsHandler outputHandler;
    private final StreamsHandler inputHandler;
    private final boolean redirectErrorStream;
    private final ProcessLauncher processLauncher;
    private int timeoutMillis;
    private boolean daemon;

    /**

		}
	}

	// We fake a local and remote server.
	remoteHost := remote.HostName()

	// 1: Echo
	register := func(manager *Manager) {
		errFatal(manager.RegisterStreamingHandler(handlerTest, StreamHandler{
			Handle: func(ctx context.Context, payload []byte, request <-chan []byte, resp chan<- []byte) *RemoteErr {
				for in := range request {
					b := append([]byte{}, payload...)
					b = append(b, in...)

	"istio.io/istio/pkg/test/util/retry"
)

type testAdscRunServer struct{}

var StreamHandler func(stream discovery.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error

func (t *testAdscRunServer) StreamAggregatedResources(stream discovery.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error {
	return StreamHandler(stream)
}

	// Buffers sent on out can not be referenced once sent.
	StreamHandlerFn func(ctx context.Context, payload []byte, in <-chan []byte, out chan<- []byte) *RemoteErr

	// StreamHandler handles fully bidirectional streams,
	// There is flow control in both directions.
	StreamHandler struct {
		// Handle an incoming request. Initial payload is sent.
		// Additional input packets (if any) are streamed to request.

      val bodyDelayNanos = response.bodyDelayNanos
      val trailers = response.trailers
      val body = response.body
      val streamHandler = response.streamHandler
      val outFinished = (
        body == null &&
          response.pushPromises.isEmpty() &&
          streamHandler == null
      )
      val flushHeaders = body == null || bodyDelayNanos != 0L
      require(!outFinished || trailers.size == 0) {