benjamyn/any-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove sugared logger

Browse Source
This commit is contained in:
mcrakhman 2022-12-09 22:11:22 +01:00 committed by Mikhail Iudin
parent ed3d7a7720
commit c31ebe6794
No known key found for this signature in database
GPG Key ID: FAAAA8BAABDFF1C0
2 changed files with 13 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
common/commonspace/syncservice/streampool.go
View File

@ -7,6 +7,7 @@ import (
"github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace/spacesyncproto" "github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace/spacesyncproto"
"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/peer" "github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/peer"
"github.com/anytypeio/go-anytype-infrastructure-experiments/common/pkg/ocache" "github.com/anytypeio/go-anytype-infrastructure-experiments/common/pkg/ocache"
"go.uber.org/zap"
"sync" "sync"
"sync/atomic" "sync/atomic"
"time" "time"
@ -97,7 +98,7 @@ func (s *streamPool) SendSync(
delete(s.waiters, msg.ReplyId) delete(s.waiters, msg.ReplyId)
s.waitersMx.Unlock() s.waitersMx.Unlock()
log.With("replyId", msg.ReplyId).Error("time elapsed when waiting") log.With(zap.String("replyId", msg.ReplyId)).Error("time elapsed when waiting")
err = ErrSyncTimeout err = ErrSyncTimeout
case reply = <-waiter.ch: case reply = <-waiter.ch:
if !delay.Stop() { if !delay.Stop() {
@ -124,8 +125,8 @@ func (s *streamPool) SendAsync(peers []string, message *spacesyncproto.ObjectSyn
streams := getStreams() streams := getStreams()
s.Unlock() s.Unlock()
log.With("objectId", message.ObjectId). log.With(zap.String("objectId", message.ObjectId), zap.Int("peers", len(streams))).
Debugf("sending message to %d peers", len(streams)) Debug("sending message to peers")
for _, s := range streams { for _, s := range streams {
err = s.Send(message) err = s.Send(message)
} }
@ -172,8 +173,8 @@ Loop:
func (s *streamPool) BroadcastAsync(message *spacesyncproto.ObjectSyncMessage) (err error) { func (s *streamPool) BroadcastAsync(message *spacesyncproto.ObjectSyncMessage) (err error) {
streams := s.getAllStreams() streams := s.getAllStreams()
log.With("objectId", message.ObjectId). log.With(zap.String("objectId", message.ObjectId), zap.Int("peers", len(streams))).
Debugf("broadcasting message to %d peers", len(streams)) Debug("broadcasting message to peers")
for _, stream := range streams { for _, stream := range streams {
if err = stream.Send(message); err != nil { if err = stream.Send(message); err != nil {
// TODO: add logging // TODO: add logging
@ -224,7 +225,7 @@ func (s *streamPool) Close() (err error) {
} }
func (s *streamPool) readPeerLoop(peerId string, stream spacesyncproto.SpaceStream) (err error) { func (s *streamPool) readPeerLoop(peerId string, stream spacesyncproto.SpaceStream) (err error) {
log.With("peerId", peerId).Debug("reading stream from peer") log.With(zap.String("replyId", peerId)).Debug("reading stream from peer")
defer s.wg.Done() defer s.wg.Done()
limiter := make(chan struct{}, maxSimultaneousOperationsPerStream) limiter := make(chan struct{}, maxSimultaneousOperationsPerStream)
for i := 0; i < maxSimultaneousOperationsPerStream; i++ { for i := 0; i < maxSimultaneousOperationsPerStream; i++ {
@ -237,17 +238,17 @@ func (s *streamPool) readPeerLoop(peerId string, stream spacesyncproto.SpaceStre
s.messageHandler(stream.Context(), peerId, msg) s.messageHandler(stream.Context(), peerId, msg)
return return
} }
log.With("replyId", msg.ReplyId).Debug("getting message with reply id") log.With(zap.String("replyId", msg.ReplyId)).Debug("getting message with reply id")
s.waitersMx.Lock() s.waitersMx.Lock()
waiter, exists := s.waiters[msg.ReplyId] waiter, exists := s.waiters[msg.ReplyId]
if !exists { if !exists {
log.With("replyId", msg.ReplyId).Debug("reply id not exists") log.With(zap.String("replyId", msg.ReplyId)).Debug("reply id not exists")
s.waitersMx.Unlock() s.waitersMx.Unlock()
s.messageHandler(stream.Context(), peerId, msg) s.messageHandler(stream.Context(), peerId, msg)
return return
} }
log.With("replyId", msg.ReplyId).Debug("reply id exists") log.With(zap.String("replyId", msg.ReplyId)).Debug("reply id exists")
delete(s.waiters, msg.ReplyId) delete(s.waiters, msg.ReplyId)
s.waitersMx.Unlock() s.waitersMx.Unlock()
@ -273,7 +274,7 @@ Loop:
limiter <- struct{}{} limiter <- struct{}{}
}() }()
} }
log.With("peerId", peerId).Debug("stopped reading stream from peer") log.With(zap.String("peerId", peerId)).Debug("stopped reading stream from peer")
s.removePeer(peerId) s.removePeer(peerId)
return return
} }

4
common/commonspace/syncservice/syncservice.go
View File

@ -13,7 +13,7 @@ import (
"time" "time"
) )
var log = logger.NewNamed("syncservice").Sugar() var log = logger.NewNamed("commonspace.syncservice")
type SyncService interface { type SyncService interface {
ocache.ObjectLastUsage ocache.ObjectLastUsage
@ -47,7 +47,7 @@ func NewSyncService(
return syncService.HandleMessage(ctx, senderId, message) return syncService.HandleMessage(ctx, senderId, message)
}) })
clientFactory := spacesyncproto.ClientFactoryFunc(spacesyncproto.NewDRPCSpaceClient) clientFactory := spacesyncproto.ClientFactoryFunc(spacesyncproto.NewDRPCSpaceClient)
syncLog := log.Desugar().With(zap.String("id", spaceId)) syncLog := log.With(zap.String("id", spaceId))
syncCtx, cancel := context.WithCancel(context.Background()) syncCtx, cancel := context.WithCancel(context.Background())
checker := NewStreamChecker( checker := NewStreamChecker(
spaceId, spaceId,