93 lines
1.8 KiB
Go
93 lines
1.8 KiB
Go
package pipe
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
func TestPipe(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
runCh := make(chan struct{})
|
|
f := func(ctx context.Context) (interface{}, error) {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
case <-runCh:
|
|
return "res0", nil
|
|
}
|
|
}
|
|
|
|
waitSignal := make(chan struct{}, 10)
|
|
signalled := 0
|
|
signal := func() {
|
|
signalled++
|
|
waitSignal <- struct{}{}
|
|
}
|
|
|
|
p, start := NewWithFunction(f)
|
|
p.OnSendCompletion = signal
|
|
go start()
|
|
require.Equal(t, false, p.Receiver.Receive())
|
|
|
|
st := p.Receiver.Status()
|
|
require.Equal(t, st.Completed, false)
|
|
require.Equal(t, st.Canceled, false)
|
|
require.Nil(t, st.Value)
|
|
require.Equal(t, signalled, 0)
|
|
|
|
close(runCh)
|
|
<-waitSignal
|
|
|
|
p.Receiver.Receive()
|
|
st = p.Receiver.Status()
|
|
require.Equal(t, st.Completed, true)
|
|
require.Equal(t, st.Canceled, false)
|
|
require.NoError(t, st.Err)
|
|
require.Equal(t, st.Value.(string), "res0")
|
|
}
|
|
|
|
func TestPipeCancel(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
runCh := make(chan struct{})
|
|
f := func(ctx context.Context) (interface{}, error) {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
case <-runCh:
|
|
return "res0", nil
|
|
}
|
|
}
|
|
|
|
waitSignal := make(chan struct{}, 10)
|
|
signalled := 0
|
|
signal := func() {
|
|
signalled++
|
|
waitSignal <- struct{}{}
|
|
}
|
|
|
|
p, start := NewWithFunction(f)
|
|
p.OnSendCompletion = signal
|
|
go start()
|
|
p.Receiver.Receive()
|
|
|
|
st := p.Receiver.Status()
|
|
require.Equal(t, st.Completed, false)
|
|
require.Equal(t, st.Canceled, false)
|
|
require.Nil(t, st.Value)
|
|
require.Equal(t, signalled, 0)
|
|
|
|
p.Receiver.Cancel()
|
|
<-waitSignal
|
|
|
|
p.Receiver.Receive()
|
|
st = p.Receiver.Status()
|
|
require.Equal(t, st.Completed, true)
|
|
require.Equal(t, st.Canceled, true)
|
|
require.Error(t, st.Err)
|
|
require.Equal(t, st.Err, context.Canceled)
|
|
}
|