Merge pull request #6 from Funfun/publish-timeout-config

make publish timeout configurable

pubzap.go

@@ -29,24 +29,42 @@ func init() {
 	}
 }
 
+// registerSink
+// url format:
+// host and path is a topic name
+// query:
+// - publishTimeout=X where X A duration string is a possibly signed
+//    sequence of decimal numbers, each with optional fraction and a unit suffix,
+//    such as "300ms", "-1.5h" or "2h45m".
+//    Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".
 func registerSink(protocol string) error {
 	return zap.RegisterSink(protocol, func(u *url.URL) (zap.Sink, error) {
 		topicName := path.Join(u.Host, u.Path)
 
+		publishTimeout := defaultPublishTimeout
+		publishTimeoutRaw := u.Query().Get("publishTimeout")
+		if publishTimeoutRaw != "" {
+			var err error
+			publishTimeout, err = time.ParseDuration(publishTimeoutRaw)
+			if err != nil {
+				return nil, err
+			}
+		}
 		ctx := context.Background()
 		topic, err := pubsub.OpenTopic(ctx, fmt.Sprintf("%s://%s", protocol, topicName))
 		if err != nil {
 			return nil, err
 		}
 
-		return &pubsubSink{topic: topic}, nil
+		return &pubsubSink{topic: topic, publishTimeout: publishTimeout}, nil
 	})
 }
 
 // pubsubSink is struct that satisfies zap.Sink
 type pubsubSink struct {
 	topic *pubsub.Topic
 
+	publishTimeout time.Duration
 	zapcore.WriteSyncer
 	io.Closer
 }
@@ -59,7 +77,7 @@ func (zpb *pubsubSink) Close() error {
 // Write implement zap.Sink func Write
 func (zpb *pubsubSink) Write(b []byte) (int, error) {
 	go func() {
-		ctx, cancel := context.WithTimeout(context.Background(), defaultPublishTimeout)
+		ctx, cancel := context.WithTimeout(context.Background(), zpb.publishTimeout)
 		defer cancel()
 
 		err := zpb.topic.Send(ctx, &pubsub.Message{

pubzap_test.go

@@ -41,7 +41,7 @@ func TestPublishAndRecieveLogs(t *testing.T) {
 	}
 	defer sub.Shutdown(ctx)
 
-	topicInfo, close, err := zap.Open("mem://topicA")
+	topicInfo, close, err := zap.Open("mem://topicA?publishTimeout=10s")
 	defer close()
 	if err != nil {
 		t.Fatal(err)