Merge pull request #27 from anyproto/change-add-to-tryadd-streampool

Cancel on cache close

Makefile

@@ -25,7 +25,6 @@ deps:
 	go mod download
 	go build -o deps storj.io/drpc/cmd/protoc-gen-go-drpc
 	go build -o deps github.com/gogo/protobuf/protoc-gen-gogofaster
-	go build -o deps github.com/golang/mock/mockgen
 
 test:
 	go test ./... --cover

commonspace/object/tree/synctree/treesyncprotocol.go

@@ -2,6 +2,7 @@ package synctree
 
 import (
 	"context"
+
 	"github.com/anyproto/any-sync/app/logger"
 	"github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"

metric/drpc.go

@@ -2,8 +2,10 @@ package metric
 
 import (
 	"github.com/prometheus/client_golang/prometheus"
+	"go.uber.org/zap"
 	"storj.io/drpc"
 	"time"
+	"unicode/utf8"
 )
 
 type prometheusDRPC struct {
@@ -14,7 +16,11 @@ type prometheusDRPC struct {
 func (ph *prometheusDRPC) HandleRPC(stream drpc.Stream, rpc string) (err error) {
 	st := time.Now()
 	defer func() {
-		ph.SummaryVec.WithLabelValues(rpc).Observe(time.Since(st).Seconds())
+		if utf8.ValidString(rpc) {
+			ph.SummaryVec.WithLabelValues(rpc).Observe(time.Since(st).Seconds())
+		} else {
+			log.WarnCtx(stream.Context(), "invalid rpc string", zap.String("rpc", rpc))
+		}
 	}()
 	return ph.Handler.HandleRPC(stream, rpc)
 }

net/streampool/stream.go

@@ -2,11 +2,12 @@ package streampool
 
 import (
 	"context"
+	"sync/atomic"
+
 	"github.com/anyproto/any-sync/app/logger"
 	"github.com/cheggaaa/mb/v3"
 	"go.uber.org/zap"
 	"storj.io/drpc"
-	"sync/atomic"
 )
 
 type stream struct {
@@ -22,7 +23,7 @@ type stream struct {
 }
 
 func (sr *stream) write(msg drpc.Message) (err error) {
-	return sr.queue.Add(sr.stream.Context(), msg)
+	return sr.queue.TryAdd(msg)
 }
 
 func (sr *stream) readLoop() error {

net/streampool/streampool_test.go

@@ -2,18 +2,19 @@ package streampool
 
 import (
 	"fmt"
+	"sort"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
 	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/net/rpc/rpctest"
 	"github.com/anyproto/any-sync/net/streampool/testservice"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/context"
-	"sort"
 	"storj.io/drpc"
-	"sync"
-	"sync/atomic"
-	"testing"
-	"time"
 )
 
 var ctx = context.Background()