merge client/server connection and stream
parent
439e39a688
commit
c87800ed83
@ -1,589 +1,154 @@
|
||||
package nghttp2
|
||||
|
||||
/*
|
||||
#cgo pkg-config: libnghttp2
|
||||
#include "_nghttp2.h"
|
||||
*/
|
||||
import "C"
|
||||
import (
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
"unsafe"
|
||||
)
|
||||
|
||||
var (
|
||||
errAgain = errors.New("again")
|
||||
)
|
||||
|
||||
// ClientConn http2 client connection
|
||||
type ClientConn struct {
|
||||
session *C.nghttp2_session
|
||||
conn net.Conn
|
||||
streams map[int]*ClientStream
|
||||
lock *sync.Mutex
|
||||
errch chan struct{}
|
||||
exitch chan struct{}
|
||||
err error
|
||||
closed bool
|
||||
streamCount int
|
||||
}
|
||||
|
||||
// Client create http2 client
|
||||
func Client(c net.Conn) (*ClientConn, error) {
|
||||
conn := &ClientConn{
|
||||
conn: c, streams: make(map[int]*ClientStream),
|
||||
lock: new(sync.Mutex),
|
||||
errch: make(chan struct{}),
|
||||
exitch: make(chan struct{}),
|
||||
}
|
||||
conn.session = C.init_nghttp2_client_session(
|
||||
C.size_t(int(uintptr(unsafe.Pointer(conn)))))
|
||||
if conn.session == nil {
|
||||
return nil, fmt.Errorf("init session failed")
|
||||
}
|
||||
ret := C.nghttp2_submit_settings(conn.session, 0, nil, 0)
|
||||
if int(ret) < 0 {
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("submit settings error: %s",
|
||||
C.GoString(C.nghttp2_strerror(ret)))
|
||||
}
|
||||
go conn.run()
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
// Error return current error on connection
|
||||
func (c *ClientConn) Error() error {
|
||||
c.lock.Lock()
|
||||
defer c.lock.Unlock()
|
||||
return c.err
|
||||
}
|
||||
|
||||
// Close close the http2 connection
|
||||
func (c *ClientConn) Close() error {
|
||||
if c.closed {
|
||||
return nil
|
||||
}
|
||||
c.closed = true
|
||||
|
||||
for _, s := range c.streams {
|
||||
s.Close()
|
||||
}
|
||||
|
||||
c.lock.Lock()
|
||||
defer c.lock.Unlock()
|
||||
//log.Println("close client connection")
|
||||
C.nghttp2_session_terminate_session(c.session, 0)
|
||||
C.nghttp2_session_del(c.session)
|
||||
close(c.exitch)
|
||||
c.conn.Close()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *ClientConn) run() {
|
||||
var wantWrite int
|
||||
var delay = 50 * time.Millisecond
|
||||
var keepalive = 5 * time.Second
|
||||
var ret C.int
|
||||
var lastDataRecv time.Time
|
||||
|
||||
//defer c.Close()
|
||||
|
||||
defer close(c.errch)
|
||||
|
||||
errch := make(chan struct{}, 5)
|
||||
|
||||
// data read loop
|
||||
go func() {
|
||||
buf := make([]byte, 16*1024)
|
||||
readloop:
|
||||
for {
|
||||
select {
|
||||
case <-c.exitch:
|
||||
break readloop
|
||||
case <-errch:
|
||||
break readloop
|
||||
default:
|
||||
}
|
||||
|
||||
n, err := c.conn.Read(buf)
|
||||
if err != nil {
|
||||
c.lock.Lock()
|
||||
c.err = err
|
||||
c.lock.Unlock()
|
||||
close(errch)
|
||||
break
|
||||
}
|
||||
//log.Printf("read %d bytes from network", n)
|
||||
lastDataRecv = time.Now()
|
||||
//d1 := C.CBytes(buf[:n])
|
||||
|
||||
c.lock.Lock()
|
||||
//ret1 := C.nghttp2_session_mem_recv(c.session,
|
||||
// (*C.uchar)(d1), C.size_t(n))
|
||||
ret1 := C.nghttp2_session_mem_recv(c.session,
|
||||
(*C.uchar)(unsafe.Pointer(&buf[0])), C.size_t(n))
|
||||
c.lock.Unlock()
|
||||
|
||||
//C.free(d1)
|
||||
if int(ret1) < 0 {
|
||||
c.lock.Lock()
|
||||
c.err = fmt.Errorf("sesion recv error: %s",
|
||||
C.GoString(C.nghttp2_strerror(ret)))
|
||||
//log.Println(c.err)
|
||||
c.lock.Unlock()
|
||||
close(errch)
|
||||
break
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// keep alive loop
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case <-c.exitch:
|
||||
return
|
||||
case <-errch:
|
||||
return
|
||||
case <-time.After(keepalive):
|
||||
}
|
||||
now := time.Now()
|
||||
last := lastDataRecv
|
||||
d := now.Sub(last)
|
||||
if d > keepalive {
|
||||
c.lock.Lock()
|
||||
C.nghttp2_submit_ping(c.session, 0, nil)
|
||||
c.lock.Unlock()
|
||||
//log.Println("submit ping")
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
loop:
|
||||
for {
|
||||
select {
|
||||
case <-c.errch:
|
||||
break loop
|
||||
case <-errch:
|
||||
break loop
|
||||
case <-c.exitch:
|
||||
break loop
|
||||
default:
|
||||
}
|
||||
|
||||
c.lock.Lock()
|
||||
ret = C.nghttp2_session_send(c.session)
|
||||
c.lock.Unlock()
|
||||
|
||||
if int(ret) < 0 {
|
||||
c.lock.Lock()
|
||||
c.err = fmt.Errorf("sesion send error: %s",
|
||||
C.GoString(C.nghttp2_strerror(ret)))
|
||||
c.lock.Unlock()
|
||||
//log.Println(c.err)
|
||||
close(errch)
|
||||
break
|
||||
}
|
||||
|
||||
c.lock.Lock()
|
||||
wantWrite = int(C.nghttp2_session_want_write(c.session))
|
||||
c.lock.Unlock()
|
||||
|
||||
// make delay when no data read/write
|
||||
if wantWrite == 0 {
|
||||
time.Sleep(delay)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Connect submit a CONNECT request, return a ClientStream and http status code from server
|
||||
//
|
||||
// equals to "CONNECT host:port" in http/1.1
|
||||
func (c *ClientConn) Connect(addr string) (cs *ClientStream, statusCode int, err error) {
|
||||
if c.err != nil {
|
||||
return nil, http.StatusServiceUnavailable, c.err
|
||||
}
|
||||
var nv = []C.nghttp2_nv{}
|
||||
nv = append(nv, newNV(":method", "CONNECT"))
|
||||
nv = append(nv, newNV(":authority", addr))
|
||||
|
||||
var dp *dataProvider
|
||||
|
||||
s := &ClientStream{
|
||||
conn: c,
|
||||
cdp: C.nghttp2_data_provider{},
|
||||
resch: make(chan *http.Response),
|
||||
errch: make(chan error),
|
||||
lock: new(sync.Mutex),
|
||||
}
|
||||
|
||||
dp = newDataProvider(unsafe.Pointer(&s.cdp), c.lock, 1)
|
||||
s.dp = dp
|
||||
|
||||
c.lock.Lock()
|
||||
streamID := C._nghttp2_submit_request(c.session, nil,
|
||||
C.size_t(uintptr(unsafe.Pointer(&nv[0]))), C.size_t(len(nv)), &s.cdp, nil)
|
||||
c.lock.Unlock()
|
||||
|
||||
if int(streamID) < 0 {
|
||||
return nil, http.StatusServiceUnavailable, fmt.Errorf(
|
||||
"submit request error: %s", C.GoString(C.nghttp2_strerror(streamID)))
|
||||
}
|
||||
dp.streamID = int(streamID)
|
||||
dp.session = c.session
|
||||
s.streamID = int(streamID)
|
||||
//log.Println("stream id ", int(streamID))
|
||||
|
||||
c.lock.Lock()
|
||||
c.streams[int(streamID)] = s
|
||||
c.streamCount++
|
||||
c.lock.Unlock()
|
||||
|
||||
//log.Printf("new stream id %d", int(streamID))
|
||||
select {
|
||||
case err := <-s.errch:
|
||||
//log.Println("wait response, got ", err)
|
||||
return nil, http.StatusServiceUnavailable, err
|
||||
case res := <-s.resch:
|
||||
if res != nil {
|
||||
res.Request = &http.Request{
|
||||
Method: "CONNECT",
|
||||
RequestURI: addr,
|
||||
URL: &url.URL{},
|
||||
Host: addr,
|
||||
}
|
||||
return s, res.StatusCode, nil
|
||||
}
|
||||
//log.Println("wait response, empty response")
|
||||
return nil, http.StatusServiceUnavailable, io.EOF
|
||||
case <-c.errch:
|
||||
return nil, http.StatusServiceUnavailable, fmt.Errorf("connection error")
|
||||
}
|
||||
}
|
||||
|
||||
func newNV(name, value string) C.nghttp2_nv {
|
||||
nv := C.nghttp2_nv{}
|
||||
nameArr := make([]byte, len(name)+1)
|
||||
valueArr := make([]byte, len(value)+1)
|
||||
copy(nameArr, []byte(name))
|
||||
copy(valueArr, []byte(value))
|
||||
|
||||
nv.name = (*C.uchar)(unsafe.Pointer(&nameArr[0]))
|
||||
nv.value = (*C.uchar)(unsafe.Pointer(&valueArr[0]))
|
||||
nv.namelen = C.size_t(len(name))
|
||||
nv.valuelen = C.size_t(len(value))
|
||||
nv.flags = 0
|
||||
return nv
|
||||
}
|
||||
|
||||
// CreateRequest submit a request and return a http.Response,
|
||||
func (c *ClientConn) CreateRequest(req *http.Request) (*http.Response, error) {
|
||||
if c.err != nil {
|
||||
return nil, c.err
|
||||
}
|
||||
|
||||
nv := []C.nghttp2_nv{}
|
||||
nv = append(nv, newNV(":method", req.Method))
|
||||
nv = append(nv, newNV(":scheme", "https"))
|
||||
nv = append(nv, newNV(":authority", req.Host))
|
||||
|
||||
/*
|
||||
:path must starts with "/"
|
||||
req.RequestURI maybe starts with http://
|
||||
*/
|
||||
p := req.URL.Path
|
||||
q := req.URL.Query().Encode()
|
||||
if q != "" {
|
||||
p = p + "?" + q
|
||||
}
|
||||
|
||||
nv = append(nv, newNV(":path", p))
|
||||
|
||||
//log.Printf("%s http://%s%s", req.Method, req.Host, p)
|
||||
for k, v := range req.Header {
|
||||
//log.Printf("header %s: %s\n", k, v[0])
|
||||
_k := strings.ToLower(k)
|
||||
if _k == "host" || _k == "connection" || _k == "proxy-connection" {
|
||||
continue
|
||||
}
|
||||
//log.Printf("header %s: %s", k, v)
|
||||
nv = append(nv, newNV(k, v[0]))
|
||||
}
|
||||
|
||||
var dp *dataProvider
|
||||
|
||||
s := &ClientStream{
|
||||
//streamID: int(streamID),
|
||||
conn: c,
|
||||
resch: make(chan *http.Response),
|
||||
errch: make(chan error),
|
||||
lock: new(sync.Mutex),
|
||||
}
|
||||
|
||||
if req.Method == "PUT" || req.Method == "POST" || req.Method == "CONNECT" {
|
||||
s.cdp = C.nghttp2_data_provider{}
|
||||
dp = newDataProvider(unsafe.Pointer(&s.cdp), c.lock, 1)
|
||||
s.dp = dp
|
||||
go func() {
|
||||
io.Copy(dp, req.Body)
|
||||
dp.Close()
|
||||
}()
|
||||
}
|
||||
|
||||
c.lock.Lock()
|
||||
streamID := C._nghttp2_submit_request(c.session, nil,
|
||||
C.size_t(uintptr(unsafe.Pointer(&nv[0]))), C.size_t(len(nv)), &s.cdp, nil)
|
||||
c.lock.Unlock()
|
||||
|
||||
//C.delete_nv_array(nva)
|
||||
|
||||
if int(streamID) < 0 {
|
||||
return nil, fmt.Errorf("submit request error: %s",
|
||||
C.GoString(C.nghttp2_strerror(streamID)))
|
||||
}
|
||||
s.streamID = int(streamID)
|
||||
//log.Printf("new stream, id %d", int(streamID))
|
||||
|
||||
if dp != nil {
|
||||
dp.streamID = int(streamID)
|
||||
dp.session = c.session
|
||||
}
|
||||
|
||||
c.lock.Lock()
|
||||
c.streams[int(streamID)] = s
|
||||
c.streamCount++
|
||||
c.lock.Unlock()
|
||||
|
||||
// waiting for response from server
|
||||
select {
|
||||
case err := <-s.errch:
|
||||
return nil, err
|
||||
case res := <-s.resch:
|
||||
if res != nil {
|
||||
res.Request = req
|
||||
return res, nil
|
||||
}
|
||||
return nil, io.EOF
|
||||
case <-c.errch:
|
||||
return nil, fmt.Errorf("connection error")
|
||||
}
|
||||
//return nil, fmt.Errorf("unknown error")
|
||||
}
|
||||
|
||||
// CanTakeNewRequest check if the ClientConn can submit a new request
|
||||
func (c *ClientConn) CanTakeNewRequest() bool {
|
||||
c.lock.Lock()
|
||||
c.lock.Unlock()
|
||||
|
||||
if c.closed {
|
||||
return false
|
||||
}
|
||||
|
||||
if c.err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
if c.streamCount > ((1 << 31) / 2) {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// ServerConn server connection
|
||||
type ServerConn struct {
|
||||
// Handler handler to handle request
|
||||
Handler http.Handler
|
||||
|
||||
closed bool
|
||||
session *C.nghttp2_session
|
||||
streams map[int]*ServerStream
|
||||
lock *sync.Mutex
|
||||
conn net.Conn
|
||||
errch chan struct{}
|
||||
exitch chan struct{}
|
||||
err error
|
||||
}
|
||||
|
||||
// HTTP2Handler is the http2 server handler that can co-work with standard net/http.
|
||||
//
|
||||
// usage example:
|
||||
// l, err := net.Listen("tcp", ":1222")
|
||||
// srv := &http.Server{
|
||||
// TLSConfig: &tls.Config{
|
||||
// NextProtos:[]string{"h2", "http/1.1"},
|
||||
// }
|
||||
// TLSNextProto: map[string]func(*http.Server, *tls.Conn, http.Handler){
|
||||
// "h2": nghttp2.Http2Handler
|
||||
// }
|
||||
// }
|
||||
// srv.ServeTLS(l, "server.crt", "server.key")
|
||||
func HTTP2Handler(srv *http.Server, conn *tls.Conn, handler http.Handler) {
|
||||
h2conn, err := Server(conn, handler)
|
||||
if err != nil {
|
||||
panic(err.Error())
|
||||
}
|
||||
h2conn.Run()
|
||||
}
|
||||
|
||||
// Server create new server connection
|
||||
func Server(c net.Conn, handler http.Handler) (*ServerConn, error) {
|
||||
conn := &ServerConn{
|
||||
conn: c,
|
||||
Handler: handler,
|
||||
streams: make(map[int]*ServerStream),
|
||||
lock: new(sync.Mutex),
|
||||
errch: make(chan struct{}),
|
||||
exitch: make(chan struct{}),
|
||||
}
|
||||
|
||||
conn.session = C.init_nghttp2_server_session(
|
||||
C.size_t(uintptr(unsafe.Pointer(conn))))
|
||||
if conn.session == nil {
|
||||
return nil, fmt.Errorf("init session failed")
|
||||
}
|
||||
|
||||
//log.Println("send server connection header")
|
||||
ret := C.send_server_connection_header(conn.session)
|
||||
if int(ret) < 0 {
|
||||
return nil, fmt.Errorf(C.GoString(C.nghttp2_strerror(ret)))
|
||||
//return nil, fmt.Errorf("send connection header failed")
|
||||
}
|
||||
//go conn.run()
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
func (c *ServerConn) serve(s *ServerStream) {
|
||||
var handler = c.Handler
|
||||
if c.Handler == nil {
|
||||
handler = http.DefaultServeMux
|
||||
}
|
||||
|
||||
if s.req.URL == nil {
|
||||
s.req.URL = &url.URL{}
|
||||
}
|
||||
|
||||
// call http.Handler to serve request
|
||||
handler.ServeHTTP(s, s.req)
|
||||
s.Close()
|
||||
}
|
||||
|
||||
// Close close the server connection
|
||||
func (c *ServerConn) Close() error {
|
||||
if c.closed {
|
||||
return nil
|
||||
}
|
||||
c.closed = true
|
||||
|
||||
for _, s := range c.streams {
|
||||
s.Close()
|
||||
}
|
||||
|
||||
c.lock.Lock()
|
||||
defer c.lock.Unlock()
|
||||
|
||||
C.nghttp2_session_terminate_session(c.session, 0)
|
||||
C.nghttp2_session_del(c.session)
|
||||
close(c.exitch)
|
||||
c.conn.Close()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Run run the server loop
|
||||
func (c *ServerConn) Run() {
|
||||
var wantWrite int
|
||||
var delay = 100 * time.Millisecond
|
||||
var ret C.int
|
||||
|
||||
defer c.Close()
|
||||
defer close(c.errch)
|
||||
|
||||
errch := make(chan struct{}, 5)
|
||||
|
||||
go func() {
|
||||
buf := make([]byte, 16*1024)
|
||||
readloop:
|
||||
for {
|
||||
select {
|
||||
case <-c.exitch:
|
||||
break readloop
|
||||
case <-errch:
|
||||
break readloop
|
||||
default:
|
||||
}
|
||||
|
||||
n, err := c.conn.Read(buf)
|
||||
if err != nil {
|
||||
c.lock.Lock()
|
||||
c.err = err
|
||||
c.lock.Unlock()
|
||||
close(errch)
|
||||
break
|
||||
}
|
||||
|
||||
//d1 := C.CBytes(buf[:n])
|
||||
|
||||
c.lock.Lock()
|
||||
//ret1 := C.nghttp2_session_mem_recv(c.session,
|
||||
// (*C.uchar)(d1), C.size_t(n))
|
||||
ret1 := C.nghttp2_session_mem_recv(c.session,
|
||||
(*C.uchar)(unsafe.Pointer(&buf[0])), C.size_t(n))
|
||||
c.lock.Unlock()
|
||||
|
||||
//C.free(d1)
|
||||
if int(ret1) < 0 {
|
||||
c.lock.Lock()
|
||||
c.err = fmt.Errorf("sesion recv error: %s",
|
||||
C.GoString(C.nghttp2_strerror(ret)))
|
||||
c.lock.Unlock()
|
||||
//log.Println(c.err)
|
||||
close(errch)
|
||||
break
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
loop:
|
||||
for {
|
||||
select {
|
||||
case <-c.errch:
|
||||
break loop
|
||||
case <-errch:
|
||||
break loop
|
||||
case <-c.exitch:
|
||||
break loop
|
||||
default:
|
||||
}
|
||||
|
||||
c.lock.Lock()
|
||||
ret = C.nghttp2_session_send(c.session)
|
||||
c.lock.Unlock()
|
||||
|
||||
if int(ret) < 0 {
|
||||
c.lock.Lock()
|
||||
c.err = fmt.Errorf("sesion send error: %s",
|
||||
C.GoString(C.nghttp2_strerror(ret)))
|
||||
c.lock.Unlock()
|
||||
//log.Println(c.err)
|
||||
close(errch)
|
||||
break
|
||||
}
|
||||
|
||||
c.lock.Lock()
|
||||
wantWrite = int(C.nghttp2_session_want_write(c.session))
|
||||
c.lock.Unlock()
|
||||
|
||||
// make delay when no data read/write
|
||||
if wantWrite == 0 {
|
||||
time.Sleep(delay)
|
||||
}
|
||||
}
|
||||
}
|
||||
package nghttp2
|
||||
|
||||
/*
|
||||
#cgo pkg-config: libnghttp2
|
||||
#include "_nghttp2.h"
|
||||
*/
|
||||
import "C"
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
"unsafe"
|
||||
)
|
||||
|
||||
// Conn http2 connection
|
||||
type Conn struct {
|
||||
conn net.Conn
|
||||
session *C.nghttp2_session
|
||||
streams map[int]*stream
|
||||
streamCount int
|
||||
closed bool
|
||||
isServer bool
|
||||
handler http.Handler
|
||||
lock *sync.Mutex
|
||||
err error
|
||||
errch chan error
|
||||
exitch chan struct{}
|
||||
}
|
||||
|
||||
// RoundTrip submit http request and return the response
|
||||
func (c *Conn) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||
return nil, errors.New("not implement")
|
||||
}
|
||||
|
||||
// Connect submit connect request
|
||||
func (c *Conn) Connect(addr string) (net.Conn, error) {
|
||||
return nil, errors.New("not implement")
|
||||
}
|
||||
|
||||
// Run run the event loop
|
||||
func (c *Conn) Run() {
|
||||
defer c.Close()
|
||||
|
||||
go c.readloop()
|
||||
go c.writeloop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case err := <-c.errch:
|
||||
c.err = err
|
||||
return
|
||||
case <-c.exitch:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Close close the connection
|
||||
func (c *Conn) Close() error {
|
||||
if c.closed {
|
||||
return nil
|
||||
}
|
||||
c.closed = true
|
||||
for _, s := range c.streams {
|
||||
s.Close()
|
||||
}
|
||||
close(c.exitch)
|
||||
c.conn.Close()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Conn) errorNotify(err error) {
|
||||
select {
|
||||
case c.errch <- err:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Conn) readloop() {
|
||||
type data struct {
|
||||
buf []byte
|
||||
err error
|
||||
}
|
||||
|
||||
var ret C.ssize_t
|
||||
var err error
|
||||
var d data
|
||||
|
||||
datach := make(chan data)
|
||||
|
||||
go func() {
|
||||
d1 := data{}
|
||||
var n int
|
||||
var err1 error
|
||||
for {
|
||||
buf := make([]byte, 16*1024)
|
||||
n, err1 = c.conn.Read(buf)
|
||||
d1.buf = buf[:n]
|
||||
d1.err = err1
|
||||
datach <- d1
|
||||
}
|
||||
}()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-c.exitch:
|
||||
return
|
||||
case d = <-datach:
|
||||
if d.err != nil {
|
||||
c.errorNotify(d.err)
|
||||
return
|
||||
}
|
||||
c.lock.Lock()
|
||||
ret = C.nghttp2_session_mem_recv(c.session,
|
||||
(*C.uchar)(unsafe.Pointer(&d.buf[0])), C.size_t(len(d.buf)))
|
||||
c.lock.Unlock()
|
||||
if int(ret) < 0 {
|
||||
err = fmt.Errorf("http2 recv error: %s", C.GoString(C.nghttp2_strerror(C.int(ret))))
|
||||
c.errorNotify(err)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Conn) writeloop() {
|
||||
var ret C.int
|
||||
var err error
|
||||
var delay = 50 * time.Millisecond
|
||||
for {
|
||||
select {
|
||||
case <-c.exitch:
|
||||
return
|
||||
default:
|
||||
}
|
||||
c.lock.Lock()
|
||||
ret = C.nghttp2_session_send(c.session)
|
||||
c.lock.Unlock()
|
||||
if int(ret) < 0 {
|
||||
err = fmt.Errorf("http2 send error: %s", C.GoString(C.nghttp2_strerror(C.int(ret))))
|
||||
c.errorNotify(err)
|
||||
return
|
||||
}
|
||||
c.lock.Lock()
|
||||
wantWrite := C.nghttp2_session_want_write(c.session)
|
||||
c.lock.Unlock()
|
||||
if int(wantWrite) == 0 {
|
||||
time.Sleep(delay)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,207 +1,48 @@
|
||||
package nghttp2
|
||||
|
||||
/*
|
||||
#include "_nghttp2.h"
|
||||
*/
|
||||
import "C"
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"unsafe"
|
||||
)
|
||||
|
||||
// ClientStream http2 client stream
|
||||
type ClientStream struct {
|
||||
streamID int
|
||||
conn *ClientConn
|
||||
cdp C.nghttp2_data_provider
|
||||
dp *dataProvider
|
||||
res *http.Response
|
||||
resch chan *http.Response
|
||||
errch chan error
|
||||
closed bool
|
||||
lock *sync.Mutex
|
||||
}
|
||||
|
||||
// Read read stream data
|
||||
func (s *ClientStream) Read(buf []byte) (n int, err error) {
|
||||
if s.closed || s.res == nil || s.res.Body == nil {
|
||||
return 0, io.EOF
|
||||
}
|
||||
return s.res.Body.Read(buf)
|
||||
}
|
||||
|
||||
// Write write data to stream
|
||||
func (s *ClientStream) Write(buf []byte) (n int, err error) {
|
||||
if s.closed {
|
||||
return 0, io.EOF
|
||||
}
|
||||
if s.dp != nil {
|
||||
return s.dp.Write(buf)
|
||||
}
|
||||
return 0, fmt.Errorf("empty data provider")
|
||||
}
|
||||
|
||||
// Close close the stream
|
||||
func (s *ClientStream) Close() error {
|
||||
//s.lock.Lock()
|
||||
//defer s.lock.Unlock()
|
||||
if s.closed {
|
||||
return nil
|
||||
}
|
||||
s.closed = true
|
||||
err := io.EOF
|
||||
//log.Printf("close stream %d", int(s.streamID))
|
||||
select {
|
||||
case s.errch <- err:
|
||||
default:
|
||||
}
|
||||
if s.res != nil && s.res.Body != nil {
|
||||
s.res.Body.Close()
|
||||
}
|
||||
//log.Println("close stream done")
|
||||
if s.dp != nil {
|
||||
s.dp.Close()
|
||||
}
|
||||
|
||||
if s.res != nil && s.res.Request != nil &&
|
||||
s.res.Request.Method == "CONNECT" {
|
||||
//log.Printf("send rst stream for %d", s.streamID)
|
||||
s.conn.lock.Lock()
|
||||
C.nghttp2_submit_rst_stream(s.conn.session, 0, C.int(s.streamID), 0)
|
||||
s.conn.lock.Unlock()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ServerStream server stream
|
||||
type ServerStream struct {
|
||||
streamID int
|
||||
// headers receive done
|
||||
headersDone bool
|
||||
// is stream_end flag received
|
||||
streamEnd bool
|
||||
// request
|
||||
req *http.Request
|
||||
// response header
|
||||
header http.Header
|
||||
// response statusCode
|
||||
statusCode int
|
||||
// response has send
|
||||
responseSend bool
|
||||
|
||||
// server connection
|
||||
conn *ServerConn
|
||||
|
||||
// data provider
|
||||
dp *dataProvider
|
||||
cdp C.nghttp2_data_provider
|
||||
|
||||
closed bool
|
||||
//buf *bytes.Buffer
|
||||
}
|
||||
|
||||
// Write write data to stream,
|
||||
// implements http.ResponseWriter
|
||||
func (s *ServerStream) Write(buf []byte) (int, error) {
|
||||
if s.closed {
|
||||
return 0, io.EOF
|
||||
}
|
||||
|
||||
if !s.responseSend {
|
||||
s.WriteHeader(http.StatusOK)
|
||||
}
|
||||
return s.dp.Write(buf)
|
||||
}
|
||||
|
||||
// WriteHeader set response code and send reponse,
|
||||
// implements http.ResponseWriter
|
||||
func (s *ServerStream) WriteHeader(code int) {
|
||||
if s.closed {
|
||||
return
|
||||
}
|
||||
|
||||
if s.responseSend {
|
||||
return
|
||||
}
|
||||
|
||||
s.responseSend = true
|
||||
s.statusCode = code
|
||||
|
||||
var nv = []C.nghttp2_nv{}
|
||||
|
||||
nv = append(nv, newNV(":status", fmt.Sprintf("%d", code)))
|
||||
|
||||
for k, v := range s.header {
|
||||
//log.Println(k, v[0])
|
||||
_k := strings.ToLower(k)
|
||||
if _k == "host" || _k == "connection" || _k == "proxy-connection" ||
|
||||
_k == "transfer-encoding" {
|
||||
continue
|
||||
}
|
||||
nv = append(nv, newNV(k, v[0]))
|
||||
}
|
||||
|
||||
var dp *dataProvider
|
||||
|
||||
dp = newDataProvider(unsafe.Pointer(&s.cdp), s.conn.lock, 0)
|
||||
dp.streamID = s.streamID
|
||||
dp.session = s.conn.session
|
||||
|
||||
s.dp = dp
|
||||
|
||||
s.conn.lock.Lock()
|
||||
ret := C._nghttp2_submit_response(
|
||||
s.conn.session, C.int(s.streamID),
|
||||
C.size_t(uintptr(unsafe.Pointer(&nv[0]))),
|
||||
C.size_t(len(nv)), &s.cdp)
|
||||
s.conn.lock.Unlock()
|
||||
|
||||
if int(ret) < 0 {
|
||||
panic(fmt.Sprintf("sumit response error %s",
|
||||
C.GoString(C.nghttp2_strerror(ret))))
|
||||
}
|
||||
//log.Printf("stream %d send response", s.streamID)
|
||||
}
|
||||
|
||||
// Header return the http.Header,
|
||||
// implements http.ResponseWriter
|
||||
func (s *ServerStream) Header() http.Header {
|
||||
if s.header == nil {
|
||||
s.header = http.Header{}
|
||||
}
|
||||
return s.header
|
||||
}
|
||||
|
||||
// Close close the stream
|
||||
func (s *ServerStream) Close() error {
|
||||
if s.closed {
|
||||
return nil
|
||||
}
|
||||
s.closed = true
|
||||
|
||||
if s.req.Body != nil {
|
||||
s.req.Body.Close()
|
||||
}
|
||||
|
||||
if s.dp != nil {
|
||||
s.dp.Close()
|
||||
//s.dp = nil
|
||||
}
|
||||
|
||||
if s.req.Method == "CONNECT" {
|
||||
s.conn.lock.Lock()
|
||||
s.conn.lock.Unlock()
|
||||
|
||||
if _, ok := s.conn.streams[s.streamID]; ok {
|
||||
//log.Printf("send rst stream %d", s.streamID)
|
||||
C.nghttp2_submit_rst_stream(s.conn.session, 0, C.int(s.streamID), 0)
|
||||
delete(s.conn.streams, s.streamID)
|
||||
}
|
||||
}
|
||||
//log.Printf("stream %d closed", s.streamID)
|
||||
return nil
|
||||
}
|
||||
package nghttp2
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net"
|
||||
"net/http"
|
||||
"time"
|
||||
)
|
||||
|
||||
type stream struct {
|
||||
streamID int
|
||||
conn *Conn
|
||||
dp *dataProvider
|
||||
bp *bodyProvider
|
||||
request *http.Request
|
||||
response *http.Response
|
||||
resch chan *http.Response
|
||||
}
|
||||
|
||||
var _ net.Conn = &stream{}
|
||||
|
||||
func (s *stream) Read(buf []byte) (int, error) {
|
||||
return 0, errors.New("not implement")
|
||||
}
|
||||
func (s *stream) Write(buf []byte) (int, error) {
|
||||
if s.conn.isServer {
|
||||
return 0, errors.New("not implement")
|
||||
}
|
||||
return 0, errors.New("not implement")
|
||||
}
|
||||
func (s *stream) Close() error {
|
||||
return nil
|
||||
}
|
||||
func (s *stream) LocalAddr() net.Addr {
|
||||
return nil
|
||||
}
|
||||
func (s *stream) RemoteAddr() net.Addr {
|
||||
return nil
|
||||
}
|
||||
func (s *stream) SetDeadline(t time.Time) error {
|
||||
return errors.New("not implement")
|
||||
}
|
||||
func (s *stream) SetReadDeadline(t time.Time) error {
|
||||
return errors.New("not implement")
|
||||
}
|
||||
func (s *stream) SetWriteDeadline(t time.Time) error {
|
||||
return errors.New("not implement")
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue