Function TestParallelShortConnection

connection_test.go:708–758

(t *testing.T)

}

func TestParallelShortConnection(t *testing.T) {
	ln := createTestTCPListener(t)
	defer ln.Close()
	address := ln.Addr().String()

	var received int64
	el, err := NewEventLoop(func(ctx context.Context, connection Connection) error {
		data, err := connection.Reader().Next(connection.Reader().Len())
		atomic.AddInt64(&received, int64(len(data)))
		if err != nil {
			return err
		}
		// t.Logf("conn[%s] received: %d, active: %v", connection.RemoteAddr(), len(data), connection.IsActive())
		return nil
	})
	MustNil(t, err)
	go func() {
		el.Serve(ln)
	}()
	defer el.Shutdown(context.Background())

	conns := 100
	sizePerConn := 1024
	totalSize := conns * sizePerConn
	var wg sync.WaitGroup
	for i := 0; i < conns; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			conn, err := DialConnection("tcp", address, time.Second)
			MustNil(t, err)
			n, err := conn.Writer().WriteBinary(make([]byte, sizePerConn))
			MustNil(t, err)
			MustTrue(t, n == sizePerConn)
			err = conn.Writer().Flush()
			MustNil(t, err)
			err = conn.Close()
			MustNil(t, err)
		}()
	}
	wg.Wait()

	t0 := time.Now()
	for atomic.LoadInt64(&received) < int64(totalSize) {
		time.Sleep(time.Millisecond)
		if time.Since(t0) > 100*time.Millisecond { // max wait 100ms
			break
		}
	}
	Equal(t, atomic.LoadInt64(&received), int64(totalSize))
}

func TestConnectionServerClose(t *testing.T) {
	ln := createTestTCPListener(t)