fangdingjun
/
nghttp2-go
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Compare commits

base: fangdingjun:merge_conn
Branches Tags
fangdingjun:master
fangdingjun:merge_conn
..
compare: fangdingjun:master
Branches Tags
fangdingjun:master
fangdingjun:merge_conn

17 Commits
merge_conn ... master

Author SHA1 Message Date
fangdingjun d785811f22 fix race condition on r/w s.request 6 years ago
fangdingjun 26af20b705 add nghttp2.Transport 6 years ago
fangdingjun f23c78ede5 use context to manage conn/stream close 6 years ago
fangdingjun 2696323ff3 go fmt 6 years ago
fangdingjun 2f09b48a89 fix crash issue on stream.Write 6 years ago
fangdingjun 0a79b2882d remove buffer pool 6 years ago
fangdingjun 8288f2a989 change conn.closed flag by hold the lock 6 years ago
fangdingjun ac31db180c fix race condition on map operation 6 years ago
fangdingjun e722eded24 use conn.Close instead of send value to channel 6 years ago
fangdingjun a360f286b0 update doc 6 years ago
fangdingjun d4ee3e4458 update README 6 years ago
fangdingjun 77b17501c3 use Pool to manage byte slice 6 years ago
fangdingjun af990484d4 not pass byte slice through channel on read loop 6 years ago
fangdingjun 462ab85e7f add finallizer function to release memory 6 years ago
fangdingjun 60510df289 add RemoteAddr and URL on Request 6 years ago
fangdingjun a1577f80df support nil cfg 6 years ago
fangdingjun 714e73abf1 add Dial function 6 years ago
7 changed files with 928 additions and 688 deletions
Show Stats

22
README.md
Unescape Escape View File

@ -10,7 +10,7 @@ server usage example:
cert, err := tls.LoadX509KeyPair("testdata/server.crt", "testdata/server.key") cert, err := tls.LoadX509KeyPair("testdata/server.crt", "testdata/server.key")
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
l, err := tls.Listen("tcp", "127.0.0.1:1100", &tls.Config{ l, err := tls.Listen("tcp", "127.0.0.1:1100", &tls.Config{
@ -18,7 +18,7 @@ server usage example:
NextProtos: []string{"h2"}, NextProtos: []string{"h2"},
}) })
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
defer l.Close() defer l.Close()
addr := l.Addr().String() addr := l.Addr().String()
@ -30,7 +30,7 @@ server usage example:
hdr.Set("aa", "bb") hdr.Set("aa", "bb")
d, err := ioutil.ReadAll(r.Body) d, err := ioutil.ReadAll(r.Body)
if err != nil { if err != nil {
ln(err) log.Println(err)
return return
} }
w.Write(d) w.Write(d)
@ -43,7 +43,7 @@ server usage example:
} }
h2conn, err := NewServerConn(c, nil) h2conn, err := NewServerConn(c, nil)
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
f("%+v", h2conn) f("%+v", h2conn)
go h2conn.Run() go h2conn.Run()
@ -57,18 +57,18 @@ client usage example:
ServerName: "nghttp2.org", ServerName: "nghttp2.org",
}) })
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
defer conn.Close() defer conn.Close()
cstate := conn.ConnectionState() cstate := conn.ConnectionState()
if cstate.NegotiatedProtocol != "h2" { if cstate.NegotiatedProtocol != "h2" {
t.Fatal("no http2 on server") log.Fatal("no http2 on server")
} }
h2conn, err := NewClientConn(conn) h2conn, err := NewClientConn(conn)
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
param := url.Values{} param := url.Values{}
@ -79,18 +79,18 @@ client usage example:
"https://nghttp2.org/httpbin/post?a=b&c=d", "https://nghttp2.org/httpbin/post?a=b&c=d",
data) data)
f("%+v", req) log.Printf("%+v", req)
req.Header.Set("user-agent", "go-nghttp2/1.0") req.Header.Set("user-agent", "go-nghttp2/1.0")
req.Header.Set("Content-Type", "application/x-www-form-urlencoded") req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
res, err := h2conn.CreateRequest(req) res, err := h2conn.CreateRequest(req)
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
if res.StatusCode != http.StatusOK { if res.StatusCode != http.StatusOK {
t.Errorf("expect %d, got %d", http.StatusOK, res.StatusCode) log.Printf("expect %d, got %d", http.StatusOK, res.StatusCode)
} }
res.Write(os.Stderr) res.Write(os.Stderr)
@ -99,7 +99,7 @@ co-work with net/http server example:
l, err := net.Listen("tcp", "127.0.0.1:1222") l, err := net.Listen("tcp", "127.0.0.1:1222")
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
srv := &http.Server{ srv := &http.Server{
TLSConfig: &tls.Config{ TLSConfig: &tls.Config{

657
callbacks.go
Unescape Escape View File

@ -1,319 +1,338 @@
package nghttp2 package nghttp2
/* /*
#include "_nghttp2.h" #include "_nghttp2.h"
*/ */
import "C" import "C"
import ( import (
"bytes" "bytes"
"crypto/tls" "context"
"errors" "crypto/tls"
"io" "errors"
"net/http" "io"
"net/url" "net/http"
"strconv" "net/url"
"strings" "runtime"
"sync" "strconv"
"unsafe" "strings"
) "sync"
"unsafe"
var ( )
errAgain = errors.New("again")
) var (
errAgain = errors.New("again")
const ( )
NGHTTP2_NO_ERROR = 0
NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE = -521 const (
NGHTTP2_ERR_CALLBACK_FAILURE = -902 NGHTTP2_NO_ERROR = 0
NGHTTP2_ERR_DEFERRED = -508 NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE = -521
) NGHTTP2_ERR_CALLBACK_FAILURE = -902
NGHTTP2_ERR_DEFERRED = -508
// onDataSourceReadCallback callback function for libnghttp2 library )
// want read data from data provider source,
// return NGHTTP2_ERR_DEFERRED will cause data frame defered, /*
// application later call nghttp2_session_resume_data will re-quene the data frame var bufPool = &sync.Pool{
// New: func() interface{} {
//export onDataSourceReadCallback return make([]byte, 16*1024)
func onDataSourceReadCallback(ptr unsafe.Pointer, streamID C.int, },
buf unsafe.Pointer, length C.size_t) C.ssize_t { }
//log.Println("onDataSourceReadCallback begin") */
conn := (*Conn)(unsafe.Pointer(uintptr(ptr)))
s, ok := conn.streams[int(streamID)] // onDataSourceReadCallback callback function for libnghttp2 library
if !ok { // want read data from data provider source,
//log.Println("client dp callback, stream not exists") // return NGHTTP2_ERR_DEFERRED will cause data frame defered,
return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE // application later call nghttp2_session_resume_data will re-quene the data frame
} //
gobuf := make([]byte, int(length)) //export onDataSourceReadCallback
n, err := s.dp.Read(gobuf) func onDataSourceReadCallback(ptr unsafe.Pointer, streamID C.int,
if err != nil { buf unsafe.Pointer, length C.size_t) C.ssize_t {
if err == io.EOF { //log.Println("onDataSourceReadCallback begin")
//log.Println("onDataSourceReadCallback end") conn := (*Conn)(unsafe.Pointer(uintptr(ptr)))
return 0 s, ok := conn.streams[int(streamID)]
} if !ok {
if err == errAgain { //log.Println("client dp callback, stream not exists")
//log.Println("onDataSourceReadCallback end") return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE
//s.dp.deferred = true }
return NGHTTP2_ERR_DEFERRED gobuf := make([]byte, int(length))
} /*
//log.Println("onDataSourceReadCallback end") _length := int(length)
return NGHTTP2_ERR_CALLBACK_FAILURE gobuf := bufPool.Get().([]byte)
} if len(gobuf) < _length {
//cbuf := C.CBytes(gobuf) gobuf = make([]byte, _length)
//defer C.free(cbuf) }
//C.memcpy(buf, cbuf, C.size_t(n)) defer bufPool.Put(gobuf)
C.memcpy(buf, unsafe.Pointer(&gobuf[0]), C.size_t(n)) */
//log.Println("onDataSourceReadCallback end")
return C.ssize_t(n) n, err := s.dp.Read(gobuf[0:])
} if err != nil {
if err == io.EOF {
// onDataChunkRecv callback function for libnghttp2 library data chunk received. //log.Println("onDataSourceReadCallback end")
// return 0
//export onDataChunkRecv }
func onDataChunkRecv(ptr unsafe.Pointer, streamID C.int, if err == errAgain {
buf unsafe.Pointer, length C.size_t) C.int { //log.Println("onDataSourceReadCallback end")
//log.Println("onDataChunkRecv begin") //s.dp.deferred = true
conn := (*Conn)(unsafe.Pointer(uintptr(ptr))) return NGHTTP2_ERR_DEFERRED
gobuf := C.GoBytes(buf, C.int(length)) }
//log.Println("onDataSourceReadCallback end")
s, ok := conn.streams[int(streamID)] return NGHTTP2_ERR_CALLBACK_FAILURE
if !ok { }
//log.Println("onDataChunkRecv end") //cbuf := C.CBytes(gobuf)
return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE //defer C.free(cbuf)
} //C.memcpy(buf, cbuf, C.size_t(n))
if s.bp == nil { C.memcpy(buf, unsafe.Pointer(&gobuf[0]), C.size_t(n))
//log.Println("empty body") //log.Println("onDataSourceReadCallback end")
//log.Println("onDataChunkRecv end") return C.ssize_t(n)
return C.int(length) }
}
//log.Println("bp write") // onDataChunkRecv callback function for libnghttp2 library data chunk received.
n, err := s.bp.Write(gobuf) //
if err != nil { //export onDataChunkRecv
return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE func onDataChunkRecv(ptr unsafe.Pointer, streamID C.int,
} buf unsafe.Pointer, length C.size_t) C.int {
//log.Println("onDataChunkRecv end") //log.Println("onDataChunkRecv begin")
return C.int(n) conn := (*Conn)(unsafe.Pointer(uintptr(ptr)))
} gobuf := C.GoBytes(buf, C.int(length))
// onDataSendCallback callback function for libnghttp2 library want send data to network. s, ok := conn.streams[int(streamID)]
// if !ok {
//export onDataSendCallback //log.Println("onDataChunkRecv end")
func onDataSendCallback(ptr unsafe.Pointer, data unsafe.Pointer, size C.size_t) C.ssize_t { return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE
//log.Println("onDataSendCallback begin") }
//log.Println("data write req ", int(size)) if s.bp == nil {
conn := (*Conn)(unsafe.Pointer(uintptr(ptr))) //log.Println("empty body")
buf := C.GoBytes(data, C.int(size)) //log.Println("onDataChunkRecv end")
//log.Println(conn.conn.RemoteAddr()) return C.int(length)
n, err := conn.conn.Write(buf) }
if err != nil { //log.Println("bp write")
//log.Println("onDataSendCallback end") n, err := s.bp.Write(gobuf)
return NGHTTP2_ERR_CALLBACK_FAILURE if err != nil {
} return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE
//log.Printf("write %d bytes to network ", n) }
//log.Println("onDataSendCallback end") //log.Println("onDataChunkRecv end")
return C.ssize_t(n) return C.int(n)
} }
// onBeginHeaderCallback callback function for begin header receive. // onDataSendCallback callback function for libnghttp2 library want send data to network.
// //
//export onBeginHeaderCallback //export onDataSendCallback
func onBeginHeaderCallback(ptr unsafe.Pointer, streamID C.int) C.int { func onDataSendCallback(ptr unsafe.Pointer, data unsafe.Pointer,
//log.Println("onBeginHeaderCallback begin") size C.size_t) C.ssize_t {
//log.Printf("stream %d begin headers", int(streamID)) //log.Println("onDataSendCallback begin")
conn := (*Conn)(unsafe.Pointer(uintptr(ptr))) //log.Println("data write req ", int(size))
conn := (*Conn)(unsafe.Pointer(uintptr(ptr)))
var TLS tls.ConnectionState buf := C.GoBytes(data, C.int(size))
if tlsconn, ok := conn.conn.(*tls.Conn); ok { //log.Println(conn.conn.RemoteAddr())
TLS = tlsconn.ConnectionState() n, err := conn.conn.Write(buf)
} if err != nil {
// client //log.Println("onDataSendCallback end")
if !conn.isServer { return NGHTTP2_ERR_CALLBACK_FAILURE
s, ok := conn.streams[int(streamID)] }
if !ok { //log.Printf("write %d bytes to network ", n)
//log.Println("onBeginHeaderCallback end") //log.Println("onDataSendCallback end")
return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE return C.ssize_t(n)
} }
s.response = &http.Response{
Proto: "HTTP/2", // onBeginHeaderCallback callback function for begin header receive.
ProtoMajor: 2, //
ProtoMinor: 0, //export onBeginHeaderCallback
Header: make(http.Header), func onBeginHeaderCallback(ptr unsafe.Pointer, streamID C.int) C.int {
Body: s.bp, //log.Println("onBeginHeaderCallback begin")
TLS: &TLS, //log.Printf("stream %d begin headers", int(streamID))
} conn := (*Conn)(unsafe.Pointer(uintptr(ptr)))
return NGHTTP2_NO_ERROR
} var TLS tls.ConnectionState
if tlsconn, ok := conn.conn.(*tls.Conn); ok {
// server TLS = tlsconn.ConnectionState()
s := &stream{ }
streamID: int(streamID), // client
conn: conn, if !conn.isServer {
bp: &bodyProvider{ s, ok := conn.streams[int(streamID)]
buf: new(bytes.Buffer), if !ok {
lock: new(sync.Mutex), //log.Println("onBeginHeaderCallback end")
}, return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE
request: &http.Request{ }
Header: make(http.Header), s.response = &http.Response{
Proto: "HTTP/2", Proto: "HTTP/2",
ProtoMajor: 2, ProtoMajor: 2,
ProtoMinor: 0, ProtoMinor: 0,
TLS: &TLS, Header: make(http.Header),
}, Body: s.bp,
} TLS: &TLS,
s.request.Body = s.bp }
return NGHTTP2_NO_ERROR
conn.streams[int(streamID)] = s }
//log.Println("onBeginHeaderCallback end") // server
return NGHTTP2_NO_ERROR s := &stream{
} streamID: int(streamID),
conn: conn,
// onHeaderCallback callback function for each header received. bp: &bodyProvider{
// buf: new(bytes.Buffer),
//export onHeaderCallback lock: new(sync.Mutex),
func onHeaderCallback(ptr unsafe.Pointer, streamID C.int, },
name unsafe.Pointer, namelen C.int, request: &http.Request{
value unsafe.Pointer, valuelen C.int) C.int { Header: make(http.Header),
//log.Println("onHeaderCallback begin") Proto: "HTTP/2",
//log.Printf("header %d", int(streamID)) ProtoMajor: 2,
conn := (*Conn)(unsafe.Pointer(uintptr(ptr))) ProtoMinor: 0,
goname := string(C.GoBytes(name, namelen)) TLS: &TLS,
govalue := string(C.GoBytes(value, valuelen)) },
}
s, ok := conn.streams[int(streamID)] s.ctx, s.cancel = context.WithCancel(context.Background())
if !ok { s.request.Body = s.bp
//log.Println("onHeaderCallback end") //log.Printf("new stream %d", int(streamID))
return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE conn.streams[int(streamID)] = s
}
var header http.Header runtime.SetFinalizer(s, (*stream).free)
if conn.isServer {
header = s.request.Header //log.Println("onBeginHeaderCallback end")
} else { return NGHTTP2_NO_ERROR
header = s.response.Header }
}
goname = strings.ToLower(goname) // onHeaderCallback callback function for each header received.
switch goname { //
case ":method": //export onHeaderCallback
s.request.Method = govalue func onHeaderCallback(ptr unsafe.Pointer, streamID C.int,
case ":scheme": name unsafe.Pointer, namelen C.int,
case ":authority": value unsafe.Pointer, valuelen C.int) C.int {
s.request.Host = govalue //log.Println("onHeaderCallback begin")
case ":path": //log.Printf("header %d", int(streamID))
s.request.RequestURI = govalue conn := (*Conn)(unsafe.Pointer(uintptr(ptr)))
u, err := url.Parse(govalue) goname := string(C.GoBytes(name, namelen))
if err != nil { govalue := string(C.GoBytes(value, valuelen))
return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE
} s, ok := conn.streams[int(streamID)]
s.request.URL = u if !ok {
case ":status": //log.Println("onHeaderCallback end")
if s.response == nil { return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE
//log.Println("empty response") }
return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE var header http.Header
} if conn.isServer {
statusCode, _ := strconv.Atoi(govalue) header = s.request.Header
s.response.StatusCode = statusCode } else {
s.response.Status = http.StatusText(statusCode) header = s.response.Header
case "content-length": }
header.Add(goname, govalue) goname = strings.ToLower(goname)
n, err := strconv.ParseInt(govalue, 10, 64) switch goname {
if err == nil { case ":method":
if conn.isServer { s.request.Method = govalue
s.request.ContentLength = n case ":scheme":
} else { case ":authority":
s.response.ContentLength = n s.request.Host = govalue
} case ":path":
} s.request.RequestURI = govalue
case "transfer-encoding": u, err := url.Parse(govalue)
header.Add(goname, govalue) if err != nil {
if conn.isServer { return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE
s.request.TransferEncoding = append(s.response.TransferEncoding, govalue) }
} else { s.request.URL = u
s.response.TransferEncoding = append(s.response.TransferEncoding, govalue) case ":status":
} if s.response == nil {
default: //log.Println("empty response")
header.Add(goname, govalue) return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE
} }
//log.Println("onHeaderCallback end") statusCode, _ := strconv.Atoi(govalue)
return NGHTTP2_NO_ERROR s.response.StatusCode = statusCode
} s.response.Status = http.StatusText(statusCode)
case "content-length":
// onHeadersDoneCallback callback function for the stream when all headers received. header.Add(goname, govalue)
// n, err := strconv.ParseInt(govalue, 10, 64)
//export onHeadersDoneCallback if err == nil {
func onHeadersDoneCallback(ptr unsafe.Pointer, streamID C.int) C.int { if conn.isServer {
//log.Println("onHeadersDoneCallback begin") s.request.ContentLength = n
//log.Printf("stream %d headers done", int(streamID)) } else {
conn := (*Conn)(unsafe.Pointer(uintptr(ptr))) s.response.ContentLength = n
s, ok := conn.streams[int(streamID)] }
if !ok { }
//log.Println("onHeadersDoneCallback end") case "transfer-encoding":
return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE header.Add(goname, govalue)
} if conn.isServer {
s.headersEnd = true s.request.TransferEncoding = append(
if conn.isServer { s.response.TransferEncoding, govalue)
if s.request.Method == "CONNECT" { } else {
go conn.serve(s) s.response.TransferEncoding = append(
} s.response.TransferEncoding, govalue)
return NGHTTP2_NO_ERROR }
} default:
select { header.Add(goname, govalue)
case s.resch <- s.response: }
default: //log.Println("onHeaderCallback end")
} return NGHTTP2_NO_ERROR
//log.Println("onHeadersDoneCallback end") }
return NGHTTP2_NO_ERROR
} // onHeadersDoneCallback callback function for the stream when all headers received.
//
// onStreamClose callback function for the stream when closed. //export onHeadersDoneCallback
// func onHeadersDoneCallback(ptr unsafe.Pointer, streamID C.int) C.int {
//export onStreamClose //log.Println("onHeadersDoneCallback begin")
func onStreamClose(ptr unsafe.Pointer, streamID C.int) C.int { //log.Printf("stream %d headers done", int(streamID))
//log.Println("onStreamClose begin") conn := (*Conn)(unsafe.Pointer(uintptr(ptr)))
//log.Printf("stream %d closed", int(streamID)) s, ok := conn.streams[int(streamID)]
conn := (*Conn)(unsafe.Pointer(uintptr(ptr))) if !ok {
//log.Println("onHeadersDoneCallback end")
stream, ok := conn.streams[int(streamID)] return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE
if ok { }
go stream.Close() s.headersEnd = true
//conn.lock.Lock() if conn.isServer {
delete(conn.streams, int(streamID)) if s.request.Method == "CONNECT" {
//go stream.Close() go conn.serve(s)
//conn.lock.Unlock() }
//log.Println("onStreamClose end") return NGHTTP2_NO_ERROR
return NGHTTP2_NO_ERROR }
} select {
//log.Println("onStreamClose end") case s.resch <- s.response:
return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE default:
} }
//log.Println("onHeadersDoneCallback end")
//export onConnectionCloseCallback return NGHTTP2_NO_ERROR
func onConnectionCloseCallback(ptr unsafe.Pointer) { }
conn := (*Conn)(unsafe.Pointer(uintptr(ptr)))
conn.err = io.EOF // onStreamClose callback function for the stream when closed.
//
// signal all goroutings exit //export onStreamClose
for i := 0; i < 4; i++ { func onStreamClose(ptr unsafe.Pointer, streamID C.int) C.int {
select { //log.Println("onStreamClose begin")
case conn.exitch <- struct{}{}: //log.Printf("stream %d closed", int(streamID))
default: conn := (*Conn)(unsafe.Pointer(uintptr(ptr)))
}
} stream, ok := conn.streams[int(streamID)]
} if ok {
go stream.Close()
//export onStreamEndCallback //log.Printf("remove stream %d", int(streamID))
func onStreamEndCallback(ptr unsafe.Pointer, streamID C.int) { //conn.lock.Lock()
conn := (*Conn)(unsafe.Pointer(uintptr(ptr))) delete(conn.streams, int(streamID))
stream, ok := conn.streams[int(streamID)] //go stream.Close()
if !ok { //conn.lock.Unlock()
return //log.Println("onStreamClose end")
} return NGHTTP2_NO_ERROR
stream.streamEnd = true }
//log.Println("onStreamClose end")
stream.bp.Close() return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE
}
if stream.conn.isServer {
if stream.request.Method != "CONNECT" { //export onConnectionCloseCallback
go conn.serve(stream) func onConnectionCloseCallback(ptr unsafe.Pointer) {
} conn := (*Conn)(unsafe.Pointer(uintptr(ptr)))
return conn.err = io.EOF
} conn.Close()
} }
//export onStreamEndCallback
func onStreamEndCallback(ptr unsafe.Pointer, streamID C.int) {
conn := (*Conn)(unsafe.Pointer(uintptr(ptr)))
stream, ok := conn.streams[int(streamID)]
if !ok {
return
}
stream.streamEnd = true
stream.bp.Close()
if stream.conn.isServer {
if stream.request.Method != "CONNECT" {
go conn.serve(stream)
}
return
}
}

85
client.go
Unescape Escape View File

@ -0,0 +1,85 @@
package nghttp2
import (
"crypto/tls"
"errors"
"fmt"
"net"
"net/http"
"strings"
"sync"
)
// Transport the nghttp2 RoundTripper implement
type Transport struct {
TLSConfig *tls.Config
DialTLS func(network, addr string, cfg *tls.Config) (*tls.Conn, error)
cacheConn map[string]*Conn
mu sync.Mutex
}
// RoundTrip send req and get res
func (tr *Transport) RoundTrip(req *http.Request) (res *http.Response, err error) {
h2conn, err := tr.getConn(req)
if err != nil {
return nil, err
}
return h2conn.RoundTrip(req)
}
func (tr *Transport) getConn(req *http.Request) (*Conn, error) {
tr.mu.Lock()
defer tr.mu.Unlock()
if tr.cacheConn == nil {
tr.cacheConn = map[string]*Conn{}
}
k := req.URL.Host
if c, ok := tr.cacheConn[k]; ok {
if c.CanTakeNewRequest() {
return c, nil
}
delete(tr.cacheConn, k)
c.Close()
}
c, err := tr.createConn(k)
if err == nil {
tr.cacheConn[k] = c
}
return c, err
}
func (tr *Transport) createConn(host string) (*Conn, error) {
dial := tls.Dial
if tr.DialTLS != nil {
dial = tr.DialTLS
}
cfg := tr.TLSConfig
if cfg == nil {
h, _, err := net.SplitHostPort(host)
if err != nil {
h = host
}
cfg = &tls.Config{
ServerName: h,
NextProtos: []string{"h2"},
}
}
if !strings.Contains(host, ":") {
host = fmt.Sprintf("%s:443", host)
}
conn, err := dial("tcp", host, cfg)
if err != nil {
return nil, err
}
if err = conn.Handshake(); err != nil {
return nil, err
}
state := conn.ConnectionState()
if state.NegotiatedProtocol != "h2" {
conn.Close()
return nil, errors.New("http2 is not supported")
}
return Client(conn)
}

216
conn.go
Unescape Escape View File

@ -7,6 +7,7 @@ package nghttp2
import "C" import "C"
import ( import (
"bytes" "bytes"
"context"
"crypto/tls" "crypto/tls"
"errors" "errors"
"fmt" "fmt"
@ -14,6 +15,7 @@ import (
"net" "net"
"net/http" "net/http"
"net/url" "net/url"
"runtime"
"strings" "strings"
"sync" "sync"
"time" "time"
@ -26,56 +28,104 @@ type Conn struct {
session *C.nghttp2_session session *C.nghttp2_session
streams map[int]*stream streams map[int]*stream
streamCount int streamCount int
closed bool
isServer bool isServer bool
running bool running bool
handler http.Handler handler http.Handler
lock *sync.Mutex lock *sync.Mutex
err error err error
errch chan error errch chan error
exitch chan struct{} ctx context.Context
cancel context.CancelFunc
} }
// Server create server side http2 connection // Dial connect to addr and create a http2 client Conn
//
// the Conn.Run have already called, should not call it again
func Dial(network, addr string, cfg *tls.Config) (*Conn, error) {
nextProto := []string{"h2"}
if cfg == nil {
_addr := addr
h, _, err := net.SplitHostPort(addr)
if err == nil {
_addr = h
}
cfg = &tls.Config{ServerName: _addr}
}
cfg.NextProtos = nextProto
conn, err := tls.Dial(network, addr, cfg)
if err != nil {
return nil, err
}
if err := conn.Handshake(); err != nil {
return nil, err
}
state := conn.ConnectionState()
if state.NegotiatedProtocol != "h2" {
return nil, errors.New("server not support http2")
}
return Client(conn)
}
// Server create server side http2 connection on c
//
// c must be TLS connection and negotiated for h2
//
// the Conn.Run not called, you must run it
func Server(c net.Conn, handler http.Handler) (*Conn, error) { func Server(c net.Conn, handler http.Handler) (*Conn, error) {
conn := &Conn{ conn := &Conn{
conn: c, conn: c,
handler: handler, handler: handler,
errch: make(chan error), errch: make(chan error),
exitch: make(chan struct{}),
lock: new(sync.Mutex), lock: new(sync.Mutex),
isServer: true, isServer: true,
streams: make(map[int]*stream), streams: make(map[int]*stream),
} }
conn.session = C.init_nghttp2_server_session(C.size_t(uintptr(unsafe.Pointer(conn))))
conn.ctx, conn.cancel = context.WithCancel(context.Background())
//log.Printf("new conn %x", uintptr(unsafe.Pointer(conn)))
runtime.SetFinalizer(conn, (*Conn).free)
conn.session = C.init_nghttp2_server_session(
C.size_t(uintptr(unsafe.Pointer(conn))))
if conn.session == nil { if conn.session == nil {
return nil, errors.New("init server session failed") return nil, errors.New("init server session failed")
} }
ret := C.send_connection_header(conn.session) ret := C.send_connection_header(conn.session)
if int(ret) < 0 { if int(ret) < 0 {
conn.Close() conn.Close()
return nil, fmt.Errorf("send settings error: %s", C.GoString(C.nghttp2_strerror(ret))) return nil, fmt.Errorf("send settings error: %s",
C.GoString(C.nghttp2_strerror(ret)))
} }
return conn, nil return conn, nil
} }
// Client create client side http2 connection // Client create client side http2 connection on c
//
// c must be TLS connection and negotiated for h2
//
// the Conn.Run have alread called, you should not call it again
func Client(c net.Conn) (*Conn, error) { func Client(c net.Conn) (*Conn, error) {
conn := &Conn{ conn := &Conn{
conn: c, conn: c,
errch: make(chan error), errch: make(chan error),
exitch: make(chan struct{}),
lock: new(sync.Mutex), lock: new(sync.Mutex),
streams: make(map[int]*stream), streams: make(map[int]*stream),
} }
conn.session = C.init_nghttp2_client_session(C.size_t(uintptr(unsafe.Pointer(conn))))
conn.ctx, conn.cancel = context.WithCancel(context.Background())
//log.Printf("new conn %x", uintptr(unsafe.Pointer(conn)))
runtime.SetFinalizer(conn, (*Conn).free)
conn.session = C.init_nghttp2_client_session(
C.size_t(uintptr(unsafe.Pointer(conn))))
if conn.session == nil { if conn.session == nil {
return nil, errors.New("init server session failed") return nil, errors.New("init server session failed")
} }
ret := C.send_connection_header(conn.session) ret := C.send_connection_header(conn.session)
if int(ret) < 0 { if int(ret) < 0 {
conn.Close() conn.Close()
return nil, fmt.Errorf("send settings error: %s", C.GoString(C.nghttp2_strerror(ret))) return nil, fmt.Errorf("send settings error: %s",
C.GoString(C.nghttp2_strerror(ret)))
} }
go conn.Run() go conn.Run()
return conn, nil return conn, nil
@ -102,6 +152,13 @@ func HTTP2Handler(srv *http.Server, conn *tls.Conn, handler http.Handler) {
h2conn.Run() h2conn.Run()
} }
func (c *Conn) free() {
//log.Printf("free conn %x", uintptr(unsafe.Pointer(c)))
if !c.isClosed() {
c.Close()
}
}
// Error return conn error // Error return conn error
func (c *Conn) Error() error { func (c *Conn) Error() error {
c.lock.Lock() c.lock.Lock()
@ -136,7 +193,8 @@ func (c *Conn) RoundTrip(req *http.Request) (*http.Response, error) {
nv = append(nv, newNV(":path", p)) nv = append(nv, newNV(":path", p))
for k, v := range req.Header { for k, v := range req.Header {
_k := strings.ToLower(k) _k := strings.ToLower(k)
if _k == "connection" || _k == "proxy-connection" || _k == "transfer-encoding" { if _k == "connection" || _k == "proxy-connection" ||
_k == "transfer-encoding" {
continue continue
} }
nv = append(nv, newNV(k, v[0])) nv = append(nv, newNV(k, v[0]))
@ -165,6 +223,9 @@ func (c *Conn) RoundTrip(req *http.Request) (*http.Response, error) {
dp.Close() dp.Close()
}() }()
} }
s.request = req
select { select {
case res := <-s.resch: case res := <-s.resch:
/* /*
@ -172,23 +233,25 @@ func (c *Conn) RoundTrip(req *http.Request) (*http.Response, error) {
return nil, fmt.Errorf("http error code %d", res.StatusCode) return nil, fmt.Errorf("http error code %d", res.StatusCode)
} }
*/ */
s.request = req
res.Request = s.request res.Request = s.request
return res, nil return res, nil
case <-c.exitch: case <-c.ctx.Done():
return nil, errors.New("connection closed") return nil, errors.New("connection closed")
} }
} }
func (c *Conn) submitRequest(nv []C.nghttp2_nv, cdp *C.nghttp2_data_provider) (*stream, error) { func (c *Conn) submitRequest(nv []C.nghttp2_nv,
cdp *C.nghttp2_data_provider) (*stream, error) {
c.lock.Lock() c.lock.Lock()
ret := C._nghttp2_submit_request(c.session, nil, ret := C._nghttp2_submit_request(c.session, nil,
C.size_t(uintptr(unsafe.Pointer(&nv[0]))), C.size_t(len(nv)), cdp, nil) C.size_t(uintptr(unsafe.Pointer(&nv[0]))),
C.size_t(len(nv)), cdp, nil)
c.lock.Unlock() c.lock.Unlock()
if int(ret) < 0 { if int(ret) < 0 {
return nil, fmt.Errorf("submit request error: %s", C.GoString(C.nghttp2_strerror(ret))) return nil, fmt.Errorf("submit request error: %s",
C.GoString(C.nghttp2_strerror(ret)))
} }
streamID := int(ret) streamID := int(ret)
s := &stream{ s := &stream{
@ -200,14 +263,22 @@ func (c *Conn) submitRequest(nv []C.nghttp2_nv, cdp *C.nghttp2_data_provider) (*
}, },
resch: make(chan *http.Response), resch: make(chan *http.Response),
} }
s.ctx, s.cancel = context.WithCancel(context.Background())
if cdp != nil { if cdp != nil {
s.cdp = *cdp s.cdp = *cdp
} }
runtime.SetFinalizer(s, (*stream).free)
return s, nil return s, nil
} }
// Connect submit connect request // Connect submit connect request
func (c *Conn) Connect(addr string) (net.Conn, int, error) { //
// like "CONNECT host:port" on http/1.1
//
// statusCode is the http status code the server returned
//
// c bounds to the remote host of addr
func (c *Conn) Connect(addr string) (conn net.Conn, statusCode int, err error) {
nv := []C.nghttp2_nv{} nv := []C.nghttp2_nv{}
nv = append(nv, newNV(":method", "CONNECT")) nv = append(nv, newNV(":method", "CONNECT"))
@ -232,7 +303,8 @@ func (c *Conn) Connect(addr string) (net.Conn, int, error) {
select { select {
case res := <-s.resch: case res := <-s.resch:
if res.StatusCode != http.StatusOK { if res.StatusCode != http.StatusOK {
return nil, res.StatusCode, fmt.Errorf("http error code %d", res.StatusCode) return nil, res.StatusCode, fmt.Errorf(
"http error code %d", res.StatusCode)
} }
s.request = &http.Request{ s.request = &http.Request{
Method: "CONNECT", Method: "CONNECT",
@ -242,8 +314,9 @@ func (c *Conn) Connect(addr string) (net.Conn, int, error) {
} }
res.Request = s.request res.Request = s.request
return s, res.StatusCode, nil return s, res.StatusCode, nil
case <-c.exitch: case <-c.ctx.Done():
return nil, http.StatusServiceUnavailable, errors.New("connection closed") return nil, http.StatusServiceUnavailable,
errors.New("connection closed")
} }
} }
@ -264,7 +337,7 @@ func (c *Conn) Run() {
case err := <-c.errch: case err := <-c.errch:
c.err = err c.err = err
return return
case <-c.exitch: case <-c.ctx.Done():
return return
} }
} }
@ -275,26 +348,40 @@ func (c *Conn) serve(s *stream) {
if handler == nil { if handler == nil {
handler = http.DefaultServeMux handler = http.DefaultServeMux
} }
s.request.RemoteAddr = c.conn.RemoteAddr().String()
if s.request.URL == nil {
s.request.URL = &url.URL{}
}
handler.ServeHTTP(s, s.request) handler.ServeHTTP(s, s.request)
s.Close() s.Close()
} }
// Close close the connection // Close close the connection
func (c *Conn) Close() error { func (c *Conn) Close() error {
if c.closed { c.lock.Lock()
if c.isClosed() {
c.lock.Unlock()
return nil return nil
} }
c.closed = true c.cancel()
c.lock.Unlock()
// stream.Close may require the conn.Lock
// so must not hold the lock here
for _, s := range c.streams { for _, s := range c.streams {
s.Close() s.Close()
} }
c.lock.Lock() c.lock.Lock()
for n := range c.streams {
delete(c.streams, n)
}
C.nghttp2_session_terminate_session(c.session, 0) C.nghttp2_session_terminate_session(c.session, 0)
C.nghttp2_session_del(c.session) C.nghttp2_session_del(c.session)
c.lock.Unlock() c.lock.Unlock()
close(c.exitch)
c.conn.Close() c.conn.Close()
return nil return nil
} }
@ -307,48 +394,31 @@ func (c *Conn) errorNotify(err error) {
} }
func (c *Conn) readloop() { func (c *Conn) readloop() {
type data struct { buf := make([]byte, 16*1024)
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 { for {
select { if c.isClosed() {
case <-c.exitch:
return return
case d = <-datach: }
if d.err != nil { n, err := c.conn.Read(buf)
c.errorNotify(d.err) if err != nil {
return c.errorNotify(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.Lock()
// check again
if c.isClosed() {
c.lock.Unlock() c.lock.Unlock()
if int(ret) < 0 { return
err = fmt.Errorf("http2 recv error: %s", C.GoString(C.nghttp2_strerror(C.int(ret)))) }
c.errorNotify(err) ret := C.nghttp2_session_mem_recv(c.session,
return (*C.uchar)(unsafe.Pointer(&buf[0])), C.size_t(n))
} 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
} }
} }
} }
@ -357,17 +427,18 @@ func (c *Conn) writeloop() {
var ret C.int var ret C.int
var err error var err error
var delay = 50 * time.Millisecond var delay = 50 * time.Millisecond
for { for {
select { c.lock.Lock()
case <-c.exitch: if c.isClosed() {
c.lock.Unlock()
return return
default:
} }
c.lock.Lock()
ret = C.nghttp2_session_send(c.session) ret = C.nghttp2_session_send(c.session)
c.lock.Unlock() c.lock.Unlock()
if int(ret) < 0 { if int(ret) < 0 {
err = fmt.Errorf("http2 send error: %s", C.GoString(C.nghttp2_strerror(C.int(ret)))) err = fmt.Errorf("http2 send error: %s",
C.GoString(C.nghttp2_strerror(C.int(ret))))
c.errorNotify(err) c.errorNotify(err)
return return
} }
@ -380,3 +451,12 @@ func (c *Conn) writeloop() {
} }
} }
} }
func (c *Conn) isClosed() bool {
select {
case <-c.ctx.Done():
return true
default:
}
return false
}

353
data_provider.go
Unescape Escape View File

@ -1,164 +1,189 @@
package nghttp2 package nghttp2
/* /*
#include "_nghttp2.h" #include "_nghttp2.h"
*/ */
import "C" import "C"
import ( import (
"bytes" "bytes"
"errors" "errors"
"log" "io"
"sync" "log"
"time" "sync"
"unsafe" "time"
) "unsafe"
)
// dataProvider provider data for libnghttp2 library
// libnghttp2 callback will Read to read the data, // dataProvider provider data for libnghttp2 library
// application call Write to provider data, // libnghttp2 callback will Read to read the data,
// application call Close will cause Read return io.EOF // application call Write to provider data,
type dataProvider struct { // application call Close will cause Read return io.EOF
buf *bytes.Buffer type dataProvider struct {
closed bool buf *bytes.Buffer
lock *sync.Mutex closed bool
sessLock *sync.Mutex lock *sync.Mutex
session *C.nghttp2_session sessLock *sync.Mutex
streamID int session *C.nghttp2_session
deferred bool streamID int
} deferred bool
}
// Read read from data provider
func (dp *dataProvider) Read(buf []byte) (n int, err error) { // Read read from data provider
if dp.buf == nil || dp.lock == nil || dp.sessLock == nil || dp.session == nil { func (dp *dataProvider) Read(buf []byte) (n int, err error) {
log.Println("dp read invalid state") if dp.buf == nil || dp.lock == nil ||
return 0, errors.New("invalid state") dp.sessLock == nil || dp.session == nil {
} log.Println("dp read invalid state")
dp.lock.Lock() return 0, errors.New("invalid state")
defer dp.lock.Unlock() }
dp.lock.Lock()
n, err = dp.buf.Read(buf) defer dp.lock.Unlock()
if err != nil && !dp.closed {
//log.Println("deferred") n, err = dp.buf.Read(buf)
dp.deferred = true if err != nil && !dp.closed {
return 0, errAgain //log.Println("deferred")
} dp.deferred = true
return return 0, errAgain
} }
return
// Write provider data for data provider }
func (dp *dataProvider) Write(buf []byte) (n int, err error) {
if dp.buf == nil || dp.lock == nil || dp.sessLock == nil || dp.session == nil { // Write provider data for data provider
log.Println("dp write invalid state") func (dp *dataProvider) Write(buf []byte) (n int, err error) {
return 0, errors.New("invalid state") if dp.buf == nil || dp.lock == nil ||
} dp.sessLock == nil || dp.session == nil {
dp.lock.Lock() log.Println("dp write invalid state")
defer dp.lock.Unlock() return 0, errors.New("invalid state")
}
//if dp.closed {
// return 0, io.EOF // make sure the buffer not too large
//} delay := 10 * time.Millisecond
maxBufSize := 4 * 1024
n, err = dp.buf.Write(buf) for {
if dp.deferred { dp.lock.Lock()
dp.sessLock.Lock() _len := dp.buf.Len()
C.nghttp2_session_resume_data(dp.session, C.int(dp.streamID)) closed := dp.closed
dp.sessLock.Unlock() dp.lock.Unlock()
if closed {
//log.Println("resume") return 0, io.EOF
dp.deferred = false }
} if _len < maxBufSize {
return break
} }
time.Sleep(delay)
// Close end to provide data }
func (dp *dataProvider) Close() error {
if dp.buf == nil || dp.lock == nil || dp.sessLock == nil || dp.session == nil { dp.lock.Lock()
log.Println("dp close, invalid state") defer dp.lock.Unlock()
return errors.New("invalid state")
} //if dp.closed {
dp.lock.Lock() // return 0, io.EOF
defer dp.lock.Unlock() //}
if dp.closed { n, err = dp.buf.Write(buf)
return nil if dp.deferred {
} dp.sessLock.Lock()
dp.closed = true C.nghttp2_session_resume_data(
//log.Printf("dp close stream %d", dp.streamID) dp.session, C.int(dp.streamID))
if dp.deferred { dp.sessLock.Unlock()
dp.sessLock.Lock()
C.nghttp2_session_resume_data(dp.session, C.int(dp.streamID)) //log.Println("resume")
dp.sessLock.Unlock() dp.deferred = false
}
dp.deferred = false return
} }
return nil
} // Close end to provide data
func (dp *dataProvider) Close() error {
func newDataProvider(cdp unsafe.Pointer, sessionLock *sync.Mutex, t int) *dataProvider { if dp.buf == nil || dp.lock == nil ||
dp := &dataProvider{ dp.sessLock == nil || dp.session == nil {
buf: new(bytes.Buffer), log.Println("dp close, invalid state")
lock: new(sync.Mutex), return errors.New("invalid state")
sessLock: sessionLock, }
} dp.lock.Lock()
C.data_provider_set_callback(C.size_t(uintptr(cdp)), defer dp.lock.Unlock()
C.size_t(uintptr(unsafe.Pointer(dp))), C.int(t))
return dp if dp.closed {
} return nil
}
// bodyProvider provide data for http body dp.closed = true
// Read will block when data not yet avaliable //log.Printf("dp close stream %d", dp.streamID)
type bodyProvider struct { if dp.deferred {
buf *bytes.Buffer dp.sessLock.Lock()
closed bool C.nghttp2_session_resume_data(
lock *sync.Mutex dp.session, C.int(dp.streamID))
} dp.sessLock.Unlock()
// Read read data from provider dp.deferred = false
// will block when data not yet avaliable }
func (bp *bodyProvider) Read(buf []byte) (int, error) { return nil
var delay = 100 * time.Millisecond }
for { func newDataProvider(cdp unsafe.Pointer,
bp.lock.Lock() sessionLock *sync.Mutex, t int) *dataProvider {
n, err := bp.buf.Read(buf) dp := &dataProvider{
bp.lock.Unlock() buf: new(bytes.Buffer),
if err != nil && !bp.closed { lock: new(sync.Mutex),
time.Sleep(delay) sessLock: sessionLock,
continue }
} C.data_provider_set_callback(C.size_t(uintptr(cdp)),
return n, err C.size_t(uintptr(unsafe.Pointer(dp))), C.int(t))
} return dp
} }
// Write provide data for dataProvider // bodyProvider provide data for http body
// libnghttp2 data chunk recv callback will call this // Read will block when data not yet avaliable
func (bp *bodyProvider) Write(buf []byte) (int, error) { type bodyProvider struct {
bp.lock.Lock() buf *bytes.Buffer
defer bp.lock.Unlock() closed bool
lock *sync.Mutex
return bp.buf.Write(buf) }
}
// Read read data from provider
// Close end to provide data // will block when data not yet avaliable
func (bp *bodyProvider) Close() error { func (bp *bodyProvider) Read(buf []byte) (int, error) {
bp.lock.Lock() var delay = 100 * time.Millisecond
defer bp.lock.Unlock()
for {
bp.closed = true bp.lock.Lock()
return nil n, err := bp.buf.Read(buf)
} bp.lock.Unlock()
if err != nil && !bp.closed {
func newNV(name, value string) C.nghttp2_nv { time.Sleep(delay)
nv := C.nghttp2_nv{} continue
nameArr := make([]byte, len(name)+1) }
valueArr := make([]byte, len(value)+1) return n, err
copy(nameArr, []byte(name)) }
copy(valueArr, []byte(value)) }
nv.name = (*C.uchar)(unsafe.Pointer(&nameArr[0])) // Write provide data for dataProvider
nv.value = (*C.uchar)(unsafe.Pointer(&valueArr[0])) // libnghttp2 data chunk recv callback will call this
nv.namelen = C.size_t(len(name)) func (bp *bodyProvider) Write(buf []byte) (int, error) {
nv.valuelen = C.size_t(len(value)) bp.lock.Lock()
nv.flags = 0 defer bp.lock.Unlock()
return nv
} return bp.buf.Write(buf)
}
// Close end to provide data
func (bp *bodyProvider) Close() error {
bp.lock.Lock()
defer bp.lock.Unlock()
bp.closed = true
return nil
}
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
}

226
doc.go
Unescape Escape View File

@ -1,113 +1,113 @@
/*Package nghttp2 is libnghttp2 binding for golang. /*Package nghttp2 is libnghttp2 binding for golang.
server example server example
cert, err := tls.LoadX509KeyPair("testdata/server.crt", "testdata/server.key") cert, err := tls.LoadX509KeyPair("testdata/server.crt", "testdata/server.key")
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
l, err := tls.Listen("tcp", "127.0.0.1:1100", &tls.Config{ l, err := tls.Listen("tcp", "127.0.0.1:1100", &tls.Config{
Certificates: []tls.Certificate{cert}, Certificates: []tls.Certificate{cert},
NextProtos: []string{"h2"}, NextProtos: []string{"h2"},
}) })
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
defer l.Close() defer l.Close()
addr := l.Addr().String() addr := l.Addr().String()
http.HandleFunc("/test", func(w http.ResponseWriter, r *http.Request) { http.HandleFunc("/test", func(w http.ResponseWriter, r *http.Request) {
log.Printf("%+v", r) log.Printf("%+v", r)
hdr := w.Header() hdr := w.Header()
hdr.Set("content-type", "text/plain") hdr.Set("content-type", "text/plain")
hdr.Set("aa", "bb") hdr.Set("aa", "bb")
d, err := ioutil.ReadAll(r.Body) d, err := ioutil.ReadAll(r.Body)
if err != nil { if err != nil {
log.Println(err) log.Println(err)
return return
} }
w.Write(d) w.Write(d)
}) })
for { for {
c, err := l.Accept() c, err := l.Accept()
if err != nil { if err != nil {
break break
} }
h2conn, err := Server(c, nil) h2conn, err := Server(c, nil)
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
log.Printf("%+v", h2conn) log.Printf("%+v", h2conn)
go h2conn.Run() go h2conn.Run()
} }
client example client example
conn, err := tls.Dial("tcp", "nghttp2.org:443", &tls.Config{ conn, err := tls.Dial("tcp", "nghttp2.org:443", &tls.Config{
NextProtos: []string{"h2"}, NextProtos: []string{"h2"},
ServerName: "nghttp2.org", ServerName: "nghttp2.org",
}) })
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
defer conn.Close() defer conn.Close()
if err := conn.Handshake(); err != nil{ if err := conn.Handshake(); err != nil{
t.Fatal(err) log.Fatal(err)
} }
cstate := conn.ConnectionState() cstate := conn.ConnectionState()
if cstate.NegotiatedProtocol != "h2" { if cstate.NegotiatedProtocol != "h2" {
t.Fatal("no http2 on server") log.Fatal("no http2 on server")
} }
h2conn, err := Client(conn) h2conn, err := Client(conn)
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
param := url.Values{} param := url.Values{}
param.Add("e", "b") param.Add("e", "b")
param.Add("f", "d") param.Add("f", "d")
data := bytes.NewReader([]byte(param.Encode())) data := bytes.NewReader([]byte(param.Encode()))
req, _ := http.NewRequest("POST", req, _ := http.NewRequest("POST",
"https://nghttp2.org/httpbin/post?a=b&c=d", "https://nghttp2.org/httpbin/post?a=b&c=d",
data) data)
log.Printf("%+v", req) log.Printf("%+v", req)
req.Header.Set("user-agent", "go-nghttp2/1.0") req.Header.Set("user-agent", "go-nghttp2/1.0")
req.Header.Set("Content-Type", "application/x-www-form-urlencoded") req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
res, err := h2conn.RoundTrip(req) res, err := h2conn.RoundTrip(req)
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
if res.StatusCode != http.StatusOK { if res.StatusCode != http.StatusOK {
t.Errorf("expect %d, got %d", http.StatusOK, res.StatusCode) log.Printf("expect %d, got %d", http.StatusOK, res.StatusCode)
} }
res.Write(os.Stderr) res.Write(os.Stderr)
co-work with net/http example co-work with net/http example
l, err := net.Listen("tcp", "127.0.0.1:1222") l, err := net.Listen("tcp", "127.0.0.1:1222")
if err != nil { if err != nil {
t.Fatal(err) log.Fatal(err)
} }
srv := &http.Server{ srv := &http.Server{
TLSConfig: &tls.Config{ TLSConfig: &tls.Config{
NextProtos: []string{"h2", "http/1.1"}, NextProtos: []string{"h2", "http/1.1"},
}, },
TLSNextProto: map[string]func(*http.Server, *tls.Conn, http.Handler){ TLSNextProto: map[string]func(*http.Server, *tls.Conn, http.Handler){
"h2": nghttp2.HTTP2Handler, "h2": nghttp2.HTTP2Handler,
}, },
} }
defer srv.Close() defer srv.Close()
srv.ServeTLS(l, "testdata/server.crt", "testdata/server.key") srv.ServeTLS(l, "testdata/server.crt", "testdata/server.key")
see http2_test.go for more details see http2_test.go for more details
*/ */
package nghttp2 package nghttp2

57
stream.go
Unescape Escape View File

@ -5,6 +5,7 @@ package nghttp2
*/ */
import "C" import "C"
import ( import (
"context"
"errors" "errors"
"fmt" "fmt"
"io" "io"
@ -25,14 +26,31 @@ type stream struct {
resch chan *http.Response resch chan *http.Response
headersEnd bool headersEnd bool
streamEnd bool streamEnd bool
closed bool
cdp C.nghttp2_data_provider cdp C.nghttp2_data_provider
ctx context.Context
cancel context.CancelFunc
} }
var _ net.Conn = &stream{} var _ net.Conn = &stream{}
func (s *stream) isClosed() bool {
select {
case <-s.ctx.Done():
return true
default:
}
return false
}
func (s *stream) free() {
//log.Printf("stream free %d", s.streamID)
if !s.isClosed() {
s.Close()
}
}
func (s *stream) Read(buf []byte) (int, error) { func (s *stream) Read(buf []byte) (int, error) {
if s.closed { if s.isClosed() {
return 0, io.EOF return 0, io.EOF
} }
if s.bp != nil { if s.bp != nil {
@ -42,7 +60,7 @@ func (s *stream) Read(buf []byte) (int, error) {
} }
func (s *stream) WriteHeader(code int) { func (s *stream) WriteHeader(code int) {
if s.closed { if s.isClosed() {
return return
} }
if s.response == nil { if s.response == nil {
@ -64,24 +82,33 @@ func (s *stream) WriteHeader(code int) {
nv = append(nv, newNV(":status", fmt.Sprintf("%d", code))) nv = append(nv, newNV(":status", fmt.Sprintf("%d", code)))
for k, v := range s.response.Header { for k, v := range s.response.Header {
_k := strings.ToLower(k) _k := strings.ToLower(k)
if _k == "host" || _k == "connection" || _k == "transfer-encoding" { if _k == "host" || _k == "connection" ||
_k == "transfer-encoding" {
continue continue
} }
nv = append(nv, newNV(k, v[0])) nv = append(nv, newNV(k, v[0]))
} }
s.cdp = C.nghttp2_data_provider{} s.cdp = C.nghttp2_data_provider{}
s.dp = newDataProvider(unsafe.Pointer(&s.cdp), s.conn.lock, 0) s.dp = newDataProvider(unsafe.Pointer(&s.cdp),
s.conn.lock, 0)
s.dp.session = s.conn.session s.dp.session = s.conn.session
s.dp.streamID = s.streamID s.dp.streamID = s.streamID
s.conn.lock.Lock() s.conn.lock.Lock()
ret := C._nghttp2_submit_response(s.conn.session, C.int(s.streamID), if s.conn.isClosed() {
C.size_t(uintptr(unsafe.Pointer(&nv[0]))), C.size_t(len(nv)), &s.cdp) s.conn.lock.Unlock()
return
}
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() s.conn.lock.Unlock()
if int(ret) < 0 { if int(ret) < 0 {
panic(fmt.Sprintf("submit response error: %s", C.GoString(C.nghttp2_strerror(ret)))) panic(fmt.Sprintf("submit response error: %s",
C.GoString(C.nghttp2_strerror(ret))))
} }
} }
@ -98,10 +125,11 @@ func (s *stream) Header() http.Header {
} }
func (s *stream) Write(buf []byte) (int, error) { func (s *stream) Write(buf []byte) (int, error) {
if s.closed { if s.isClosed() {
return 0, io.EOF return 0, io.EOF
} }
if s.conn.isServer && (s.response == nil || s.response.StatusCode == 0) { if s.conn.isServer && (s.response == nil ||
s.response.StatusCode == 0) {
s.WriteHeader(http.StatusOK) s.WriteHeader(http.StatusOK)
} }
@ -112,10 +140,12 @@ func (s *stream) Write(buf []byte) (int, error) {
} }
func (s *stream) Close() error { func (s *stream) Close() error {
if s.closed { if s.isClosed() {
return nil return nil
} }
s.closed = true
s.cancel()
if s.dp != nil { if s.dp != nil {
s.dp.Close() s.dp.Close()
} }
@ -130,7 +160,8 @@ func (s *stream) Close() error {
if s.request != nil && s.request.Method == "CONNECT" { if s.request != nil && s.request.Method == "CONNECT" {
//log.Println("rst stream") //log.Println("rst stream")
s.conn.lock.Lock() s.conn.lock.Lock()
C.nghttp2_submit_rst_stream(s.conn.session, 0, C.int(s.streamID), 8) C.nghttp2_submit_rst_stream(s.conn.session, 0,
C.int(s.streamID), 8)
s.conn.lock.Unlock() s.conn.lock.Unlock()
} }
return nil return nil

Write Preview
Loading…
Cancel
Save