Stop a transcode whose request has ended before it encodes (closes #243) #271
@@ -0,0 +1,90 @@
|
|||||||
|
package imageprocessor
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"io"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// readerAfterDeadline is a Process input whose first Read waits until done
|
||||||
|
// is closed, so the deadline passes after Process has taken its slot and
|
||||||
|
// before it encodes.
|
||||||
|
type readerAfterDeadline struct {
|
||||||
|
data *bytes.Reader
|
||||||
|
done <-chan struct{}
|
||||||
|
waited bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *readerAfterDeadline) Read(p []byte) (int, error) {
|
||||||
|
if !r.waited {
|
||||||
|
r.waited = true
|
||||||
|
|
||||||
|
<-r.done
|
||||||
|
}
|
||||||
|
|
||||||
|
return r.data.Read(p)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestProcessStopsWhenDeadlinePassed checks that Process returns
|
||||||
|
// context.DeadlineExceeded, and holds no slot afterwards, when its context's
|
||||||
|
// deadline has passed before it starts or while it reads its input. libvips
|
||||||
|
// cannot stop an encode once it has begun, so Process must not begin one for
|
||||||
|
// a request that has ended.
|
||||||
|
func TestProcessStopsWhenDeadlinePassed(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
input := createTestJPEG(t, 50, 50)
|
||||||
|
|
||||||
|
t.Run("before Process starts", func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx, cancel := context.WithDeadline(t.Context(),
|
||||||
|
time.Now().Add(-time.Second))
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
checkProcessStopsAtDeadline(ctx, t, bytes.NewReader(input))
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("while Process reads its input", func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
checkProcessStopsAtDeadline(ctx, t, &readerAfterDeadline{
|
||||||
|
data: bytes.NewReader(input), done: ctx.Done(),
|
||||||
|
})
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// checkProcessStopsAtDeadline runs Process on input with ctx, whose deadline
|
||||||
|
// has passed by the time Process could encode, and fails the test unless
|
||||||
|
// Process returns context.DeadlineExceeded and gives its slot back.
|
||||||
|
func checkProcessStopsAtDeadline(
|
||||||
|
ctx context.Context, t *testing.T, input io.Reader,
|
||||||
|
) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
proc := New(Params{MaxConcurrentProcessing: 1})
|
||||||
|
|
||||||
|
result, err := proc.Process(ctx, input, smallJPEGRequest())
|
||||||
|
if result != nil {
|
||||||
|
t.Logf("Process() made a %dx%d image after its deadline passed",
|
||||||
|
result.Width, result.Height)
|
||||||
|
|
||||||
|
_ = result.Content.Close()
|
||||||
|
}
|
||||||
|
|
||||||
|
t.Logf("Process() error = %v", err)
|
||||||
|
|
||||||
|
if !errors.Is(err, context.DeadlineExceeded) {
|
||||||
|
t.Errorf("Process() error = %v, want context.DeadlineExceeded", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if held := len(proc.processingSemaphore); held != 0 {
|
||||||
|
t.Errorf("%d slots still held after Process() returned", held)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user