Merge 1b439ee into 1ac52c3

.travis.yml

@@ -2,6 +2,10 @@ language: go
 
 go:
   - 1.x
+  - 1.11.x
+
+env:
+  - GO111MODULE=on
 
 install:
   - go get golang.org/x/lint/golint

README.md

@@ -22,6 +22,8 @@
 
 ## Install
 
+> Minimal go version is ***1.11***.
+
 ```shell
 $ go get -u github.com/rsocket/rsocket-go
 ```

core/framing/frame_lease.go

@@ -101,7 +101,7 @@ func (l WriteableLeaseFrame) Len() int {
 // NewWriteableLeaseFrame creates a new WriteableLeaseFrame.
 func NewWriteableLeaseFrame(ttl time.Duration, n uint32, metadata []byte) *WriteableLeaseFrame {
 	var a, b [4]byte
-	binary.BigEndian.PutUint32(a[:], uint32(ttl.Milliseconds()))
+	binary.BigEndian.PutUint32(a[:], uint32(int64(ttl)/1e6))
 	binary.BigEndian.PutUint32(b[:], n)
 
 	var flag core.FrameFlag
@@ -121,7 +121,7 @@ func NewWriteableLeaseFrame(ttl time.Duration, n uint32, metadata []byte) *Write
 // NewLeaseFrame creates a new LeaseFrame.
 func NewLeaseFrame(ttl time.Duration, n uint32, metadata []byte) *LeaseFrame {
 	bf := common.NewByteBuff()
-	if err := binary.Write(bf, binary.BigEndian, uint32(ttl.Milliseconds())); err != nil {
+	if err := binary.Write(bf, binary.BigEndian, uint32(int64(ttl)/1e6)); err != nil {
 		panic(err)
 	}
 	if err := binary.Write(bf, binary.BigEndian, n); err != nil {

core/transport/tcp_transport.go

@@ -153,18 +153,14 @@ func NewTCPClientTransport(c net.Conn) *Transport {
 // NewTCPClientTransportWithAddr creates a new transport.
 func NewTCPClientTransportWithAddr(ctx context.Context, network, addr string, tlsConfig *tls.Config) (tp *Transport, err error) {
 	var conn net.Conn
-	if tlsConfig == nil {
-		var dial net.Dialer
-		conn, err = dial.DialContext(ctx, network, addr)
-	} else {
-		dial := tls.Dialer{
-			Config: tlsConfig,
-		}
-		conn, err = dial.DialContext(ctx, network, addr)
-	}
+	var dial net.Dialer
+	conn, err = dial.DialContext(ctx, network, addr)
 	if err != nil {
 		return
 	}
+	if tlsConfig != nil {
+		conn = tls.Client(conn, tlsConfig)
+	}
 	tp = NewTCPClientTransport(conn)
 	return
 }

core/transport/transport.go

@@ -158,7 +158,7 @@ func (p *Transport) Start(ctx context.Context) error {
 			if err == nil {
 				continue
 			}
-			if errors.Is(err, io.EOF) {
+			if err == io.EOF {
 				return nil
 			}
 			return errors.Wrap(err, "read and delivery frame failed")

examples/echo_bench/echo_bench.go

@@ -9,6 +9,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/jjeffcaii/reactor-go/scheduler"
 	"github.com/rsocket/rsocket-go"
 	"github.com/rsocket/rsocket-go/core/transport"
 	"github.com/rsocket/rsocket-go/payload"
@@ -56,7 +57,7 @@ func main() {
 		}),
 	)
 	for i := 0; i < n; i++ {
-		client.RequestResponse(payload.New(data, nil)).SubscribeWith(context.Background(), sub)
+		client.RequestResponse(payload.New(data, nil)).SubscribeOn(scheduler.Parallel()).SubscribeWith(context.Background(), sub)
 	}
 	wg.Wait()
 	cost := time.Since(now)

go.mod

@@ -1,6 +1,6 @@
 module github.com/rsocket/rsocket-go
 
-go 1.12
+go 1.11
 
 require (
 	github.com/golang/mock v1.4.3

internal/common/u24_test.go

@@ -40,7 +40,7 @@ func TestMustNewUint24(t *testing.T) {
 
 func TestUint24(t *testing.T) {
 	testSingle(t, 0)
-	for range [1_000_000]struct{}{} {
+	for range [1000000]struct{}{} {
 		testSingle(t, RandIntn(MaxUint24))
 	}
 	testSingle(t, MaxUint24)