This commit is contained in:
Asim Aslam 2018-07-29 10:55:46 +01:00
parent 88505388c1
commit eebaa64d8c

View File

@ -1,6 +1,7 @@
package transport
import (
//"fmt"
"bufio"
"bytes"
"crypto/tls"
@ -13,10 +14,11 @@ import (
"sync"
"time"
"github.com/micro/go-log"
"github.com/micro/h2c"
maddr "github.com/micro/util/go/lib/addr"
mnet "github.com/micro/util/go/lib/net"
mls "github.com/micro/util/go/lib/tls"
"golang.org/x/net/http2"
)
type buffer struct {
@ -42,12 +44,13 @@ type httpTransportClient struct {
type httpTransportSocket struct {
ht *httpTransport
r chan *http.Request
conn net.Conn
once sync.Once
w http.ResponseWriter
r *http.Request
rw *bufio.ReadWriter
sync.Mutex
buff *bufio.Reader
conn net.Conn
// for the first request
ch chan *http.Request
}
type httpTransportListener struct {
@ -175,28 +178,43 @@ func (h *httpTransportSocket) Recv(m *Message) error {
return errors.New("message passed in is nil")
}
if m.Header == nil {
m.Header = make(map[string]string)
}
// process http 1
if h.r.ProtoMajor == 1 {
// set timeout if its greater than 0
if h.ht.opts.Timeout > time.Duration(0) {
h.conn.SetDeadline(time.Now().Add(h.ht.opts.Timeout))
}
r, err := http.ReadRequest(h.buff)
var r *http.Request
select {
// get first request
case r = <-h.ch:
// read next request
default:
rr, err := http.ReadRequest(h.rw.Reader)
if err != nil {
return err
}
r = rr
}
// read body
b, err := ioutil.ReadAll(r.Body)
if err != nil {
return err
}
// set body
r.Body.Close()
m.Body = b
if m.Header == nil {
m.Header = make(map[string]string)
}
for k, v := range r.Header {
// set headers
for k, v := range h.r.Header {
if len(v) > 0 {
m.Header[k] = v[0]
} else {
@ -204,23 +222,45 @@ func (h *httpTransportSocket) Recv(m *Message) error {
}
}
select {
case h.r <- r:
default:
// return early early
return nil
}
// processing http2 request
// read streaming body
// set max buffer size
buf := make([]byte, 4*1024)
// read the request body
n, err := h.r.Body.Read(buf)
// not an eof error
if err != nil {
return err
}
// check if we have data
if n > 0 {
m.Body = buf[:n]
}
// set headers
for k, v := range h.r.Header {
if len(v) > 0 {
m.Header[k] = v[0]
} else {
m.Header[k] = ""
}
}
return nil
}
func (h *httpTransportSocket) Send(m *Message) error {
b := bytes.NewBuffer(m.Body)
defer b.Reset()
r := <-h.r
if h.r.ProtoMajor == 1 {
rsp := &http.Response{
Header: r.Header,
Body: &buffer{b},
Header: h.r.Header,
Body: ioutil.NopCloser(bytes.NewReader(m.Body)),
Status: "200 OK",
StatusCode: 200,
Proto: "HTTP/1.1",
@ -233,25 +273,31 @@ func (h *httpTransportSocket) Send(m *Message) error {
rsp.Header.Set(k, v)
}
select {
case h.r <- r:
default:
}
// set timeout if its greater than 0
if h.ht.opts.Timeout > time.Duration(0) {
h.conn.SetDeadline(time.Now().Add(h.ht.opts.Timeout))
}
return rsp.Write(h.conn)
}
// http2 request
// set headers
for k, v := range m.Header {
h.w.Header().Set(k, v)
}
// write request
_, err := h.w.Write(m.Body)
return err
}
func (h *httpTransportSocket) error(m *Message) error {
b := bytes.NewBuffer(m.Body)
defer b.Reset()
if h.r.ProtoMajor == 1 {
rsp := &http.Response{
Header: make(http.Header),
Body: &buffer{b},
Body: ioutil.NopCloser(bytes.NewReader(m.Body)),
Status: "500 Internal Server Error",
StatusCode: 500,
Proto: "HTTP/1.1",
@ -265,17 +311,15 @@ func (h *httpTransportSocket) error(m *Message) error {
}
return rsp.Write(h.conn)
}
return nil
}
func (h *httpTransportSocket) Close() error {
err := h.conn.Close()
h.once.Do(func() {
h.Lock()
h.buff.Reset(nil)
h.buff = nil
h.Unlock()
})
return err
if h.r.ProtoMajor == 1 {
return h.conn.Close()
}
return nil
}
func (h *httpTransportListener) Addr() string {
@ -287,46 +331,67 @@ func (h *httpTransportListener) Close() error {
}
func (h *httpTransportListener) Accept(fn func(Socket)) error {
var tempDelay time.Duration
// create handler mux
mux := http.NewServeMux()
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
var buf *bufio.ReadWriter
var con net.Conn
for {
c, err := h.listener.Accept()
// read a regular request
if r.ProtoMajor == 1 {
b, err := ioutil.ReadAll(r.Body)
if err != nil {
if ne, ok := err.(net.Error); ok && ne.Temporary() {
if tempDelay == 0 {
tempDelay = 5 * time.Millisecond
} else {
tempDelay *= 2
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
if max := 1 * time.Second; tempDelay > max {
tempDelay = max
}
log.Logf("http: Accept error: %v; retrying in %v\n", err, tempDelay)
time.Sleep(tempDelay)
continue
}
return err
r.Body = ioutil.NopCloser(bytes.NewReader(b))
// hijack the conn
hj, ok := w.(http.Hijacker)
if !ok {
// we're screwed
http.Error(w, "cannot serve conn", http.StatusInternalServerError)
return
}
sock := &httpTransportSocket{
conn, bufrw, err := hj.Hijack()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
defer conn.Close()
buf = bufrw
con = conn
}
// save the request
ch := make(chan *http.Request, 1)
ch <- r
fn(&httpTransportSocket{
ht: h.ht,
conn: c,
buff: bufio.NewReader(c),
r: make(chan *http.Request, 1),
w: w,
r: r,
rw: buf,
ch: ch,
conn: con,
})
})
// default http2 server
srv := &http.Server{
Handler: mux,
}
go func() {
// TODO: think of a better error response strategy
defer func() {
if r := recover(); r != nil {
log.Log("panic recovered: ", r)
sock.Close()
// insecure connection use h2c
if !(h.ht.opts.Secure || h.ht.opts.TLSConfig != nil) {
srv.Handler = &h2c.HandlerH2C{
Handler: mux,
H2Server: &http2.Server{},
}
}
}()
fn(sock)
}()
}
// begin serving
return srv.Serve(h.listener)
}
func (h *httpTransport) Dial(addr string, opts ...DialOption) (Client, error) {