/
http.go
73 lines (64 loc) · 1.49 KB
/
http.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
package protocol
import (
"bufio"
"bytes"
"context"
"fmt"
"io"
"net/http"
"net/http/httputil"
"time"
"github.com/iron-io/functions/api/models"
"github.com/iron-io/functions/api/runner/task"
)
// HTTPProtocol converts stdin/stdout streams into HTTP/1.1 compliant
// communication. It relies on Content-Length to know when to stop reading from
// containers stdout. It also mandates valid HTTP headers back and forth, thus
// returning errors in case of parsing problems.
type HTTPProtocol struct {
in io.Writer
out io.Reader
}
func (p *HTTPProtocol) IsStreamable() bool {
return true
}
func (p *HTTPProtocol) Dispatch(ctx context.Context, t task.Request) error {
var retErr error
done := make(chan struct{})
go func() {
var body bytes.Buffer
io.Copy(&body, t.Config.Stdin)
req, err := http.NewRequest("GET", "/", &body)
if err != nil {
retErr = err
return
}
for k, v := range t.Config.Env {
req.Header.Set(k, v)
}
req.Header.Set("Content-Length", fmt.Sprint(body.Len()))
req.Header.Set("Task-ID", t.Config.ID)
raw, err := httputil.DumpRequest(req, true)
if err != nil {
retErr = err
return
}
p.in.Write(raw)
res, err := http.ReadResponse(bufio.NewReader(p.out), req)
if err != nil {
retErr = err
return
}
io.Copy(t.Config.Stdout, res.Body)
done <- struct{}{}
}()
timeout := time.After(t.Config.Timeout)
select {
case <-ctx.Done():
return ctx.Err()
case <-timeout:
return models.ErrRunnerTimeout
case <-done:
return retErr
}
}