<commit_before>package frames

import (
	"bufio"
	"fmt"
	"io"
	"net"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

type testService struct {
	addr     string
	l        net.Listener
	channels int32
	msgs     int32
}

func runTestServer(t *testing.T) *testService {
	t.Parallel()
	ta, err := net.ResolveTCPAddr("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("Error resolving test server addr: %v", err)
	}
	l, err := net.ListenTCP("tcp", ta)
	if err != nil {
		t.Fatalf("Error listening: %v", err)
	}

	ll, err := ListenerListener(l)
	if err != nil {
		t.Fatalf("Error listen listening: %v", err)
	}

	t.Logf("Listening on %v", l.Addr())

	rv := testService{addr: l.Addr().String(), l: l}

	workIt := func(c net.Conn) {
		atomic.AddInt32(&rv.channels, 1)
		b := bufio.NewReader(c)
		for {
			l, err := b.ReadString('\n')
			switch err {
			case nil:
				atomic.AddInt32(&rv.msgs, 1)
			case io.EOF:
				return
			default:
				t.Errorf("Error reading: %v", err)
				t.Fail()
				return
			}
			fmt.Fprintf(c, "Ack your %v", l)
		}

	}

	go func() {
		defer ll.Close()
		for {
			c, err := ll.Accept()
			if err != nil {
				return
			}
			go workIt(c)
		}
	}()

	return &rv
}

func TestEndToEnd(t *testing.T) {
	defer time.AfterFunc(time.Millisecond*250, func() {
		panic("Taking too long")
	}).Stop()
	tc := runTestServer(t)
	defer tc.l.Close()

	c, err := net.Dial("tcp", tc.addr)
	if err != nil {
		t.Fatalf("Error connecting to my server: %v", err)
	}

	fc := NewClient(c)

	wg := sync.WaitGroup{}

	worker := func(n int, fc ChannelDialer) {
		defer wg.Done()

		c, err := fc.Dial()
		if err != nil {
			t.Fatalf("Error dialing channel: %v", err)
		}
		defer c.Close()<commit_msg>Increase test deadline for e2e test.<commit_after>package frames

import (
	"bufio"
	"fmt"
	"io"
	"net"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

type testService struct {
	addr     string
	l        net.Listener
	channels int32
	msgs     int32
}

func runTestServer(t *testing.T) *testService {
	t.Parallel()
	ta, err := net.ResolveTCPAddr("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("Error resolving test server addr: %v", err)
	}
	l, err := net.ListenTCP("tcp", ta)
	if err != nil {
		t.Fatalf("Error listening: %v", err)
	}

	ll, err := ListenerListener(l)
	if err != nil {
		t.Fatalf("Error listen listening: %v", err)
	}

	t.Logf("Listening on %v", l.Addr())

	rv := testService{addr: l.Addr().String(), l: l}

	workIt := func(c net.Conn) {
		atomic.AddInt32(&rv.channels, 1)
		b := bufio.NewReader(c)
		for {
			l, err := b.ReadString('\n')
			switch err {
			case nil:
				atomic.AddInt32(&rv.msgs, 1)
			case io.EOF:
				return
			default:
				t.Errorf("Error reading: %v", err)
				t.Fail()
				return
			}
			fmt.Fprintf(c, "Ack your %v", l)
		}

	}

	go func() {
		defer ll.Close()
		for {
			c, err := ll.Accept()
			if err != nil {
				return
			}
			go workIt(c)
		}
	}()

	return &rv
}

func TestEndToEnd(t *testing.T) {
	defer time.AfterFunc(time.Second*5, func() {
		panic("Taking too long")
	}).Stop()
	tc := runTestServer(t)
	defer tc.l.Close()

	c, err := net.Dial("tcp", tc.addr)
	if err != nil {
		t.Fatalf("Error connecting to my server: %v", err)
	}

	fc := NewClient(c)

	wg := sync.WaitGroup{}

	worker := func(n int, fc ChannelDialer) {
		defer wg.Done()

		c, err := fc.Dial()
		if err != nil {
			t.Fatalf("Error dialing channel: %v", err)
		}
		defer c.Close()
{'fraction_non_alphanumeric': 0.17652276759314015, 'fraction_numerical': 0.011531638083973979, 'mean_word_length': 2.9019607843137254, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '7074560', 'n_tokens_mistral': 1541, 'n_tokens_neox': 1393, 'n_words': 348}