<commit_before>package tcp

import (
	"fmt"
	"io"

	"v2ray.com/core/common/net"
)

type Server struct {
	Port         net.Port
	MsgProcessor func(msg []byte) []byte
	ShouldClose  bool
	SendFirst    []byte
	Listen       net.Address
	listener     *net.TCPListener
}

func (server *Server) Start() (net.Destination, error) {
	listenerAddr := server.Listen
	if listenerAddr == nil {
		listenerAddr = net.LocalHostIP
	}
	listener, err := net.ListenTCP("tcp", &net.TCPAddr{
		IP:   listenerAddr.IP(),
		Port: int(server.Port),
		Zone: "",
	})
	if err != nil {
		return net.Destination{}, err
	}
	server.Port = net.Port(listener.Addr().(*net.TCPAddr).Port)
	server.listener = listener
	go server.acceptConnections(listener)
	localAddr := listener.Addr().(*net.TCPAddr)
	return net.TCPDestination(net.IPAddress(localAddr.IP), net.Port(localAddr.Port)), nil
}

func (server *Server) acceptConnections(listener *net.TCPListener) {
	for {
		conn, err := listener.Accept()
		if err != nil {
			fmt.Printf("Failed accept TCP connection: %v\n", err)
			return
		}

		go server.handleConnection(conn)
	}
}

func (server *Server) handleConnection(conn net.Conn) {
	if len(server.SendFirst) > 0 {
		conn.Write(server.SendFirst)
	}
	request := make([]byte, 4096)
	for {
		nBytes, err := conn.Read(request)
		if err != nil {
			if err != io.EOF {
				fmt.Println("Failed to read request:", err)
			}
			break
		}
		response := server.MsgProcessor(request[:nBytes])
		if _, err := conn.Write(response); err != nil {
			fmt.Println("Failed to write response:", err)
			break
		}
		if server.ShouldClose {
			break
		}
	}
	conn.Close()
}

func (server *Server) Close() error {
	return server.listener.Close()
}<commit_msg>fix tcp testing server<commit_after>package tcp

import (
	"fmt"
	"io"

	"v2ray.com/core/common/buf"
	"v2ray.com/core/common/net"
	"v2ray.com/core/common/task"
	"v2ray.com/core/transport/pipe"
)

type Server struct {
	Port         net.Port
	MsgProcessor func(msg []byte) []byte
	ShouldClose  bool
	SendFirst    []byte
	Listen       net.Address
	listener     *net.TCPListener
}

func (server *Server) Start() (net.Destination, error) {
	listenerAddr := server.Listen
	if listenerAddr == nil {
		listenerAddr = net.LocalHostIP
	}
	listener, err := net.ListenTCP("tcp", &net.TCPAddr{
		IP:   listenerAddr.IP(),
		Port: int(server.Port),
		Zone: "",
	})
	if err != nil {
		return net.Destination{}, err
	}
	server.Port = net.Port(listener.Addr().(*net.TCPAddr).Port)
	server.listener = listener
	go server.acceptConnections(listener)
	localAddr := listener.Addr().(*net.TCPAddr)
	return net.TCPDestination(net.IPAddress(localAddr.IP), net.Port(localAddr.Port)), nil
}

func (server *Server) acceptConnections(listener *net.TCPListener) {
	for {
		conn, err := listener.Accept()
		if err != nil {
			fmt.Printf("Failed accept TCP connection: %v\n", err)
			return
		}

		go server.handleConnection(conn)
	}
}

func (server *Server) handleConnection(conn net.Conn) {
	if len(server.SendFirst) > 0 {
		conn.Write(server.SendFirst)
	}

	pReader, pWriter := pipe.New(pipe.WithoutSizeLimit())
	err := task.Run(task.Parallel(func() error {
		for {
			b := buf.New()
			if err := b.AppendSupplier(buf.ReadFrom(conn)); err != nil {
				if err == io.EOF {
					return nil
				}
				return err
			}
			copy(b.Bytes(), server.MsgProcessor(b.Bytes()))
			if err := pWriter.WriteMultiBuffer(buf.NewMultiBufferValue(b)); err != nil {
				return err
			}
		}
	}, func() error {
		w := buf.NewWriter(conn)
		for {
			mb, err := pReader.ReadMultiBuffer()
			if err != nil {
				return err
			}
			if err := w.WriteMultiBuffer(mb); err != nil {
				return err
			}
		}
	}))()

	if err != nil {
		fmt.Println("failed to transfer data: ", err.Error())
	}

	conn.Close() // nolint: errcheck
}

func (server *Server) Close() error {
	return server.listener.Close()
}
{'fraction_non_alphanumeric': 0.1516019796822089, 'fraction_numerical': 0.0028653295128939827, 'mean_word_length': 3.5283018867924527, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3071730', 'n_tokens_mistral': 1451, 'n_tokens_neox': 1345, 'n_words': 303}