fix race condition on r/w s.request

README.md

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

callbacks.go

@@ -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
+	}
+}

client.go

@@ -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)
+}

conn.go

@@ -7,6 +7,7 @@ package nghttp2
 import "C"
 import (
 	"bytes"
+	"context"
 	"crypto/tls"
 	"errors"
 	"fmt"
@@ -14,6 +15,7 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"runtime"
 	"strings"
 	"sync"
 	"time"
@@ -26,56 +28,104 @@ type Conn struct {
 	session     *C.nghttp2_session
 	streams     map[int]*stream
 	streamCount int
-	closed      bool
 	isServer    bool
 	running     bool
 	handler     http.Handler
 	lock        *sync.Mutex
 	err         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) {
 	conn := &Conn{
 		conn:     c,
 		handler:  handler,
 		errch:    make(chan error),
-		exitch:   make(chan struct{}),
 		lock:     new(sync.Mutex),
 		isServer: true,
 		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 {
 		return nil, errors.New("init server session failed")
 	}
 	ret := C.send_connection_header(conn.session)
 	if int(ret) < 0 {
 		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
 }
 
-// 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) {
 	conn := &Conn{
 		conn:    c,
 		errch:   make(chan error),
-		exitch:  make(chan struct{}),
 		lock:    new(sync.Mutex),
 		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 {
 		return nil, errors.New("init server session failed")
 	}
 	ret := C.send_connection_header(conn.session)
 	if int(ret) < 0 {
 		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()
 	return conn, nil
@@ -102,6 +152,13 @@ func HTTP2Handler(srv *http.Server, conn *tls.Conn, handler http.Handler) {
 	h2conn.Run()
 }
 
+func (c *Conn) free() {
+	//log.Printf("free conn %x", uintptr(unsafe.Pointer(c)))
+	if !c.isClosed() {
+		c.Close()
+	}
+}
+
 // Error return conn error
 func (c *Conn) Error() error {
 	c.lock.Lock()
@@ -136,7 +193,8 @@ func (c *Conn) RoundTrip(req *http.Request) (*http.Response, error) {
 	nv = append(nv, newNV(":path", p))
 	for k, v := range req.Header {
 		_k := strings.ToLower(k)
-		if _k == "connection" || _k == "proxy-connection" || _k == "transfer-encoding" {
+		if _k == "connection" || _k == "proxy-connection" ||
+			_k == "transfer-encoding" {
 			continue
 		}
 		nv = append(nv, newNV(k, v[0]))
@@ -165,6 +223,9 @@ func (c *Conn) RoundTrip(req *http.Request) (*http.Response, error) {
 			dp.Close()
 		}()
 	}
+
+	s.request = req
+
 	select {
 	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)
 			}
 		*/
-		s.request = req
 		res.Request = s.request
 		return res, nil
-	case <-c.exitch:
+	case <-c.ctx.Done():
 		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()
 	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()
 
 	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)
 	s := &stream{
@@ -200,14 +263,22 @@ func (c *Conn) submitRequest(nv []C.nghttp2_nv, cdp *C.nghttp2_data_provider) (*
 		},
 		resch: make(chan *http.Response),
 	}
+	s.ctx, s.cancel = context.WithCancel(context.Background())
 	if cdp != nil {
 		s.cdp = *cdp
 	}
+	runtime.SetFinalizer(s, (*stream).free)
 	return s, nil
 }
 
 // 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 = append(nv, newNV(":method", "CONNECT"))
@@ -232,7 +303,8 @@ func (c *Conn) Connect(addr string) (net.Conn, int, error) {
 	select {
 	case res := <-s.resch:
 		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{
 			Method:     "CONNECT",
@@ -242,8 +314,9 @@ func (c *Conn) Connect(addr string) (net.Conn, int, error) {
 		}
 		res.Request = s.request
 		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:
 			c.err = err
 			return
-		case <-c.exitch:
+		case <-c.ctx.Done():
 			return
 		}
 	}
@@ -275,26 +348,40 @@ func (c *Conn) serve(s *stream) {
 	if handler == nil {
 		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)
 	s.Close()
 }
 
 // Close close the connection
 func (c *Conn) Close() error {
-	if c.closed {
+	c.lock.Lock()
+	if c.isClosed() {
+		c.lock.Unlock()
 		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 {
 		s.Close()
 	}
 
 	c.lock.Lock()
+
+	for n := range c.streams {
+		delete(c.streams, n)
+	}
+
 	C.nghttp2_session_terminate_session(c.session, 0)
 	C.nghttp2_session_del(c.session)
 	c.lock.Unlock()
 
-	close(c.exitch)
 	c.conn.Close()
 	return nil
 }
@@ -307,48 +394,31 @@ func (c *Conn) errorNotify(err error) {
 }
 
 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 {
-		select {
+		if c.isClosed() {
-		case <-c.exitch:
 			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()
-			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 err error
 	var delay = 50 * time.Millisecond
+
 	for {
-		select {
+		c.lock.Lock()
-		case <-c.exitch:
+		if c.isClosed() {
+			c.lock.Unlock()
 			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))))
+			err = fmt.Errorf("http2 send error: %s",
+				C.GoString(C.nghttp2_strerror(C.int(ret))))
 			c.errorNotify(err)
 			return
 		}
@@ -380,3 +451,12 @@ func (c *Conn) writeloop() {
 		}
 	}
 }
+
+func (c *Conn) isClosed() bool {
+	select {
+	case <-c.ctx.Done():
+		return true
+	default:
+	}
+	return false
+}

data_provider.go

@@ -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
+}

doc.go

@@ -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

stream.go

@@ -5,6 +5,7 @@ package nghttp2
 */
 import "C"
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -25,14 +26,31 @@ type stream struct {
 	resch      chan *http.Response
 	headersEnd bool
 	streamEnd  bool
-	closed     bool
 	cdp        C.nghttp2_data_provider
+	ctx        context.Context
+	cancel     context.CancelFunc
 }
 
 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) {
-	if s.closed {
+	if s.isClosed() {
 		return 0, io.EOF
 	}
 	if s.bp != nil {
@@ -42,7 +60,7 @@ func (s *stream) Read(buf []byte) (int, error) {
 }
 
 func (s *stream) WriteHeader(code int) {
-	if s.closed {
+	if s.isClosed() {
 		return
 	}
 	if s.response == nil {
@@ -64,24 +82,33 @@ func (s *stream) WriteHeader(code int) {
 	nv = append(nv, newNV(":status", fmt.Sprintf("%d", code)))
 	for k, v := range s.response.Header {
 		_k := strings.ToLower(k)
-		if _k == "host" || _k == "connection" || _k == "transfer-encoding" {
+		if _k == "host" || _k == "connection" ||
+			_k == "transfer-encoding" {
 			continue
 		}
 		nv = append(nv, newNV(k, v[0]))
 	}
 
 	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.streamID = s.streamID
 
 	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()
 
 	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) {
-	if s.closed {
+	if s.isClosed() {
 		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)
 	}
 
@@ -112,10 +140,12 @@ func (s *stream) Write(buf []byte) (int, error) {
 }
 
 func (s *stream) Close() error {
-	if s.closed {
+	if s.isClosed() {
 		return nil
 	}
-	s.closed = true
+
+	s.cancel()
+
 	if s.dp != nil {
 		s.dp.Close()
 	}
@@ -130,7 +160,8 @@ func (s *stream) Close() error {
 	if s.request != nil && s.request.Method == "CONNECT" {
 		//log.Println("rst stream")
 		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()
 	}
 	return nil