}
	return err
}

func (f *File) writeTo(w io.Writer) (written int64, handled bool, err error) {
	pfd, network := getPollFDAndNetwork(w)
	// TODO(panjf2000): same as File.spliceToFile.
	if pfd == nil || !pfd.IsStream || !isUnixOrTCP(string(network)) {
		return
	}

	sc, err := f.SyscallConn()
	if err != nil {
		return
	}

	rerr := sc.Read(func(fd uintptr) (done bool) {

// descriptor isn't closed until all FD instances that refer to it have been
// closed/destroyed.
func (fd *FD) Copy() FD {
	return FD{
		Sysfd:         fd.Sysfd,
		SysFile:       fd.SysFile.ref(),
		IsStream:      fd.IsStream,
		ZeroReadIsEOF: fd.ZeroReadIsEOF,
		isBlocking:    fd.isBlocking,
		isFile:        fd.isFile,
	}
}

// dupCloseOnExecOld always errors on wasip1 because there is no mechanism to

	// If the request context is canceled, the stream will no longer process requests.
	// Requests sent cannot be used any further by the called.
	Requests chan<- []byte

	muxID uint64
	ctx   context.Context
}

// Send a payload to the remote server.
func (s *Stream) Send(b []byte) error {
	if s.Requests == nil {
		return errors.New("stream does not accept requests")
	}
	select {

  SP_Event event_handle_;
};

class CStream : public StreamCommon {
 public:
  CStream(SP_Device* device, SP_StreamExecutor* stream_executor,
          StreamExecutor* executor)
      : StreamCommon(executor),
        device_(device),
        stream_executor_(stream_executor),
        stream_handle_(nullptr) {}
  ~CStream() override {
    parent()->BlockHostUntilDone(this).IgnoreError();

            stream.received(bytes)
        }

        then:
        1 * dispatch.onOutput({ it instanceof ReadStdInEvent }) >> { instant.requested }
    }

    def "read byte returns when stream is closed"() {
        def dispatch = Mock(OutputEventListener)
        def stream = new StdInStream(dispatch)

        when:
        async {
            start {
                def b1 = stream.read()

TEST(OStreamTest, TestInt64) {
  Int64 x(42);
  std::stringstream stream;
  stream << x;
  ASSERT_EQ(stream.str(), "42");
}

TEST(OStreamTest, TestFloat32) {
  Float32 x(0.375);  // Exactly representable as a float.
  std::stringstream stream;
  stream << x;
  ASSERT_EQ(stream.str(), "0.375");
}

TEST(OStreamTest, TestString) {
  String s("foo");
  std::stringstream stream;
  stream << s;
  ASSERT_EQ(stream.str(), "foo");
}

        given:
        def stream = factory.newStream()
        stream.forCurrentCaller()
        stream.forCurrentCaller()

        expect:
        stream.forCurrentCaller().stack.empty

        def failure1 = new Exception("broken")
        def diagnostics1 = stream.forCurrentCaller(failure1)
        diagnostics1.exception == failure1

import org.gradle.internal.os.OperatingSystem;

import java.io.File;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class WindowsInstallationSupplier implements InstallationSupplier {

    private final WindowsRegistry windowsRegistry;
    private final OperatingSystem os;

//  done_event->Init();
//
//  XlaHostSendDeviceContext device_context(&stream, &gpu_dst,
//    shape, done_event);
//  device_context.CopyCPUTensorToDeviceSync(
//    &cpu_tensor, &device, &device_tensor);

class XlaHostSendDeviceContext : public DeviceContext {
 public:
  XlaHostSendDeviceContext(
      se::Stream* stream, se::DeviceMemoryBase* device_memory_base,

import javax.annotation.Nullable;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;

/**
 * Provides a way to decode structured data from a backing byte stream. Implementations may buffer incoming bytes read
 * from the backing stream prior to decoding.
 */
public interface Decoder {
    /**
     * Returns an InputStream which can be used to read raw bytes.
     */
    InputStream getInputStream();