func TestStreamFramer_Order_PlainText(t *testing.T) {
	// Create the stream framer
	r, w := io.Pipe()
	wrappedW := &WriteCloseChecker{WriteCloser: w}
	// Ensure the batch window doesn't get hit
	hRate, bWindow := 100*time.Millisecond, 10*time.Millisecond
	sf := NewStreamFramer(wrappedW, true, hRate, bWindow, 10)
	sf.Run()

	files := []string{"1", "2", "3", "4", "5"}
	input := bytes.NewBuffer(make([]byte, 0, 100000))
	for i := 0; i <= 1000; i++ {
		str := strconv.Itoa(i) + ","
		input.WriteString(str)
	}

	expected := bytes.NewBuffer(make([]byte, 0, 100000))
	for _, _ = range files {
		expected.Write(input.Bytes())
	}
	receivedBuf := bytes.NewBuffer(make([]byte, 0, 100000))

	// Start the reader
	resultCh := make(chan struct{})
	go func() {
	OUTER:
		for {
			if _, err := receivedBuf.ReadFrom(r); err != nil {
				if strings.Contains(err.Error(), "closed pipe") {
					resultCh <- struct{}{}
					return
				}
				t.Fatalf("bad read: %v", err)
			}

			if expected.Len() != receivedBuf.Len() {
				continue
			}
			expectedBytes := expected.Bytes()
			actualBytes := receivedBuf.Bytes()
			for i, e := range expectedBytes {
				if a := actualBytes[i]; a != e {
					continue OUTER
				}
			}
			resultCh <- struct{}{}
			return

		}
	}()

	// Send the data
	b := input.Bytes()
	shards := 10
	each := len(b) / shards
	for _, f := range files {
		for i := 0; i < shards; i++ {
			l, r := each*i, each*(i+1)
			if i == shards-1 {
				r = len(b)
			}

			if err := sf.Send(f, "", b[l:r], 0); err != nil {
				t.Fatalf("Send() failed %v", err)
			}
		}
	}

	// Ensure we get data
	select {
	case <-resultCh:
	case <-time.After(10 * time.Duration(testutil.TestMultiplier()) * bWindow):
		if expected.Len() != receivedBuf.Len() {
			t.Fatalf("Got %v; want %v", expected.Len(), receivedBuf.Len())
		}
		expectedBytes := expected.Bytes()
		actualBytes := receivedBuf.Bytes()
		for i, e := range expectedBytes {
			if a := actualBytes[i]; a != e {
				t.Fatalf("Index %d; Got %q; want %q", i, a, e)
			}
		}
	}

	// Close the reader and wait. This should cause the runner to exit
	if err := r.Close(); err != nil {
		t.Fatalf("failed to close reader")
	}

	sf.Destroy()
	if !wrappedW.Closed {
		t.Fatalf("writer not closed")
	}
}
// This test checks that frames are received in order
This test checks that frames are received in order
// This test checks that frames are received in order
func TestStreamFramer_Order_PlainText(t *testing.T) {
	r, w := io.Pipe()
	wrappedW := &WriteCloseChecker{WriteCloser: w}
	hRate, bWindow := 100*time.Millisecond, 10*time.Millisecond
	sf := NewStreamFramer(wrappedW, true, hRate, bWindow, 10)
	sf.Run()
	files := []string{"1", "2", "3", "4", "5"}
	input := bytes.NewBuffer(make([]byte, 0, 100000))
	for i := 0; i <= 1000; i++ {
		str := strconv.Itoa(i) + ","
		input.WriteString(str)
	}
	expected := bytes.NewBuffer(make([]byte, 0, 100000))
	for _, _ = range files {
		expected.Write(input.Bytes())
	}
	receivedBuf := bytes.NewBuffer(make([]byte, 0, 100000))
	resultCh := make(chan struct{})
	go func() {
	OUTER:
		for {
			if _, err := receivedBuf.ReadFrom(r); err != nil {
				if strings.Contains(err.Error(), "closed pipe") {
					resultCh <- struct{}{}
					return
				}
				t.Fatalf("bad read: %v", err)
			}
			if expected.Len() != receivedBuf.Len() {
				continue
			}
			expectedBytes := expected.Bytes()
			actualBytes := receivedBuf.Bytes()
			for i, e := range expectedBytes {
				if a := actualBytes[i]; a != e {
					continue OUTER
				}
			}
			resultCh <- struct{}{}
			return
		}
	}()
	b := input.Bytes()
	shards := 10
	each := len(b) / shards
	for _, f := range files {
		for i := 0; i < shards; i++ {
			l, r := each*i, each*(i+1)
			if i == shards-1 {
				r = len(b)
			}
			if err := sf.Send(f, "", b[l:r], 0); err != nil {
				t.Fatalf("Send() failed %v", err)
			}
		}
	}
	select {
	case <-resultCh:
	case <-time.After(10 * time.Duration(testutil.TestMultiplier()) * bWindow):
		if expected.Len() != receivedBuf.Len() {
			t.Fatalf("Got %v; want %v", expected.Len(), receivedBuf.Len())
		}
		expectedBytes := expected.Bytes()
		actualBytes := receivedBuf.Bytes()
		for i, e := range expectedBytes {
			if a := actualBytes[i]; a != e {
				t.Fatalf("Index %d; Got %q; want %q", i, a, e)
			}
		}
	}
	if err := r.Close(); err != nil {
		t.Fatalf("failed to close reader")
	}
	sf.Destroy()
	if !wrappedW.Closed {
		t.Fatalf("writer not closed")
	}
}
func TestStreamFramer_Order_PlainText ( t * testing . T ) { r , w := io . Pipe ( ) wrappedW := & WriteCloseChecker { WriteCloser : w } hRate , bWindow := 100 * time . Millisecond , 10 * time . Millisecond sf := NewStreamFramer ( wrappedW , true , hRate , bWindow , 10 ) sf . Run ( ) files := [ ] string { " " , " " , " " , " " , " " } input := bytes . NewBuffer ( make ( [ ] byte , 0 , 100000 ) ) for i := 0 ; i <= 1000 ; i ++ { str := strconv . Itoa ( i ) + " " input . WriteString ( str ) } expected := bytes . NewBuffer ( make ( [ ] byte , 0 , 100000 ) ) for _ , _ = range files { expected . Write ( input . Bytes ( ) ) } receivedBuf := bytes . NewBuffer ( make ( [ ] byte , 0 , 100000 ) ) resultCh := make ( chan struct { } ) go func ( ) { OUTER : for { if _ , err := receivedBuf . ReadFrom ( r ) ; err != nil { if strings . Contains ( err . Error ( ) , " " ) { resultCh <- struct { } { } return } t . Fatalf ( " " , err ) } if expected . Len ( ) != receivedBuf . Len ( ) { continue } expectedBytes := expected . Bytes ( ) actualBytes := receivedBuf . Bytes ( ) for i , e := range expectedBytes { if a := actualBytes [ i ] ; a != e { continue OUTER } } resultCh <- struct { } { } return } } ( ) b := input . Bytes ( ) shards := 10 each := len ( b ) / shards for _ , f := range files { for i := 0 ; i < shards ; i ++ { l , r := each * i , each * ( i + 1 ) if i == shards - 1 { r = len ( b ) } if err := sf . Send ( f , " " , b [ l : r ] , 0 ) ; err != nil { t . Fatalf ( " " , err ) } } } select { case <- resultCh : case <- time . After ( 10 * time . Duration ( testutil . TestMultiplier ( ) ) * bWindow ) : if expected . Len ( ) != receivedBuf . Len ( ) { t . Fatalf ( " " , expected . Len ( ) , receivedBuf . Len ( ) ) } expectedBytes := expected . Bytes ( ) actualBytes := receivedBuf . Bytes ( ) for i , e := range expectedBytes { if a := actualBytes [ i ] ; a != e { t . Fatalf ( " " , i , a , e ) } } } if err := r . Close ( ) ; err != nil { t . Fatalf ( " " ) } sf . Destroy ( ) if ! wrappedW . Closed { t . Fatalf ( " " ) } }
This test checks that frames are received in order
This test checks that frames are received in order
// Create the stream framer // Ensure the batch window doesn't get hit // Start the reader // Send the data // Ensure we get data // Close the reader and wait. This should cause the runner to exit