reader.go raw
1 package io
2
3 import (
4 "bufio"
5 "net"
6 "strings"
7 "time"
8
9 "git.smesh.lol/torly/pkg/httpsrv"
10 "git.smesh.lol/torly/pkg/wire"
11 "git.smesh.lol/torly/pkg/ws"
12 )
13
14 const (
15 httpReadTimeout = 30 * time.Second
16 wsIdleTimeout = 150 * time.Second
17 )
18
19 // Reader is one reader worker domain. It owns the read side and all parsing
20 // state for one connection at a time; the paired Writer owns the write side.
21 func Reader(data ReaderData, ch ReaderChans) {
22 for {
23 select {
24 case <-ch.Stop.Done():
25 return
26 default:
27 }
28 var a wire.Assign
29 select {
30 case a = <-ch.Assign.Recv():
31 case <-ch.Stop.Done():
32 return
33 }
34 if a.Socket == nil {
35 continue
36 }
37 serve(data, ch, a)
38 }
39 }
40
41 func serve(data ReaderData, ch ReaderChans, a wire.Assign) {
42 conn := a.Socket.Conn
43 if !sendCtl(ch, wire.WriterCtl{Kind: wire.CtlBind, Conn: a.ID, Socket: a.Socket}) {
44 conn.Close()
45 return
46 }
47 if !waitAck(ch, wire.CtlBind) {
48 return
49 }
50 br := bufio.NewReaderSize(conn, 16<<10)
51 maxBody := data.Cfg.BlossomMaxBlob + 4096
52 if maxBody < 1<<20 {
53 maxBody = 1 << 20
54 }
55 _ = conn.SetReadDeadline(time.Now().Add(httpReadTimeout))
56 req, err := httpsrv.ReadRequest(br, conn.RemoteAddr().String(), 32<<10, maxBody, func() error {
57 sendCtl(ch, wire.WriterCtl{Kind: wire.CtlWrite, Payload: []byte("HTTP/1.1 100 Continue\r\n\r\n")})
58 return nil
59 })
60 if err != nil || req == nil {
61 sendIn(ch, wire.Inbound{Conn: a.ID, Kind: wire.InClosed, Reason: "http read"})
62 sendCtl(ch, wire.WriterCtl{Kind: wire.CtlClose, Conn: a.ID})
63 return
64 }
65
66 if isWebSocket(req.Header) {
67 resp, herr := ws.HandshakeResponse(req.Header)
68 if herr != nil {
69 body := herr.Error() + "\n"
70 bad := wire.HTTPResponse{
71 Status: 400,
72 Header: [][2]string{{"Content-Type", "text/plain"}, {"Content-Length", itoa(len(body))}},
73 Body: []byte(body),
74 }
75 sendCtl(ch, wire.WriterCtl{Kind: wire.CtlRespond, Payload: SerializeHTTP(bad), Close: true})
76 return
77 }
78 if !sendCtl(ch, wire.WriterCtl{Kind: wire.CtlUpgrade, Conn: a.ID, Payload: resp}) {
79 return
80 }
81 if !waitAck(ch, wire.CtlUpgrade) {
82 return
83 }
84 if !sendIn(ch, wire.Inbound{Conn: a.ID, Kind: wire.InOpened}) {
85 return
86 }
87 readWebSocket(data, ch, a.ID, br, conn)
88 return
89 }
90
91 if resp, handled := localResponse(data.Cfg, req); handled {
92 sendCtl(ch, wire.WriterCtl{Kind: wire.CtlRespond, Payload: SerializeHTTP(resp), Close: true})
93 return
94 }
95
96 // Health and Blossom requests are decided by the network domain. Once the
97 // request is handed over, the reader has no further read work: the writer
98 // sends the response and closes the socket.
99 sendIn(ch, wire.Inbound{
100 Conn: a.ID,
101 Kind: wire.InHTTP,
102 HTTP: wire.HTTPRequest{
103 Method: req.Method,
104 Path: req.Path,
105 Query: req.RawQuery,
106 Header: req.Header,
107 Body: req.Body,
108 },
109 })
110 }
111
112 func readWebSocket(data ReaderData, ch ReaderChans, id wire.ConnID, br *bufio.Reader, conn net.Conn) {
113 max := int64(data.Cfg.MaxEventBytes*2 + 8192)
114 if max < 256<<10 {
115 max = 256 << 10
116 }
117 reader := ws.NewReader(br, max)
118 for {
119 select {
120 case <-ch.Stop.Done():
121 sendCtl(ch, wire.WriterCtl{Kind: wire.CtlClose, Conn: id})
122 return
123 default:
124 }
125 _ = conn.SetReadDeadline(time.Now().Add(wsIdleTimeout))
126 op, payload, err := reader.Read()
127 if err != nil {
128 sendIn(ch, wire.Inbound{Conn: id, Kind: wire.InClosed, Reason: err.Error()})
129 sendCtl(ch, wire.WriterCtl{Kind: wire.CtlClose, Conn: id})
130 return
131 }
132 switch op {
133 case ws.OpText:
134 if !sendIn(ch, wire.Inbound{Conn: id, Kind: wire.InText, Text: payload}) {
135 return
136 }
137 case ws.OpPing:
138 sendCtl(ch, wire.WriterCtl{Kind: wire.CtlPong, Conn: id, Payload: payload})
139 case ws.OpPong:
140 // nothing
141 case ws.OpClose:
142 sendIn(ch, wire.Inbound{Conn: id, Kind: wire.InClosed, Reason: "client close"})
143 sendCtl(ch, wire.WriterCtl{Kind: wire.CtlClose, Conn: id})
144 return
145 case ws.OpBinary:
146 sendCtl(ch, wire.WriterCtl{Kind: wire.CtlClose, Conn: id, Payload: closeReason(1003)})
147 sendIn(ch, wire.Inbound{Conn: id, Kind: wire.InClosed, Reason: "binary frame"})
148 return
149 }
150 }
151 }
152
153 func isWebSocket(h map[string]string) bool {
154 for _, part := range strings.Split(h["upgrade"], ",") {
155 if strings.EqualFold(strings.TrimSpace(part), "websocket") {
156 return true
157 }
158 }
159 return false
160 }
161
162 func closeReason(code uint16) []byte {
163 return ws.AppendClose(nil, code, "")
164 }
165
166 func sendIn(ch ReaderChans, m wire.Inbound) bool {
167 select {
168 case ch.In.Tx <- m:
169 return true
170 case <-ch.Stop.Done():
171 return false
172 }
173 }
174
175 func waitAck(ch ReaderChans, kind uint8) bool {
176 select {
177 case got := <-ch.Ack.Recv():
178 return got == kind
179 case <-ch.Stop.Done():
180 return false
181 }
182 }
183
184 func sendCtl(ch ReaderChans, m wire.WriterCtl) bool {
185 select {
186 case ch.Ctl.Tx <- m:
187 return true
188 case <-ch.Stop.Done():
189 return false
190 }
191 }
192
193 func itoa(n int) string {
194 if n == 0 {
195 return "0"
196 }
197 var b [20]byte
198 i := len(b)
199 for n > 0 {
200 i--
201 b[i] = byte('0' + n%10)
202 n /= 10
203 }
204 return string(b[i:])
205 }
206