Use DocRefKey type in Pubsub

admin/client.go

@@ -257,11 +257,11 @@ func (c *Client) ListDocuments(
 		ctx,
 		&api.ListDocumentsRequest{
 			ProjectName:     projectName,
+			PreviousKey:     previousOffset.Key.String(),
 			PreviousId:      previousOffset.ID.String(),
 			PageSize:        pageSize,
 			IsForward:       isForward,
 			IncludeSnapshot: includeSnapshot,
-			PreviousKey:     previousOffset.Key.String(),
 		},
 	)
 	if err != nil {

api/types/resource_ref_key.go

@@ -35,7 +35,7 @@ func (r *DocRefKey) String() string {
 	return fmt.Sprintf("Doc (%s.%s)", r.Key, r.ID)
 }
 
-// Set parses the given string (format: `docKey},{docID}`) and assigns the values
+// Set parses the given string (format: `{docKey},{docID}`) and assigns the values
 // to the given DocRefKey.
 func (r *DocRefKey) Set(v string) error {
 	parsed := strings.Split(v, ",")

cmd/yorkie/document/list.go

@@ -104,7 +104,7 @@ func init() {
 	cmd.Flags().Var(
 		&previousOffset,
 		"previous-offset",
-		"The previous document offset to start from",
+		"The previous document offset to start from. Use the format 'docKey,docID' for the input.",
 	)
 	cmd.Flags().Int32Var(
 		&pageSize,

pkg/document/change/id.go

@@ -38,7 +38,7 @@ type ID struct {
 	clientSeq uint32
 
 	// serverSeq is the sequence of the change on the server. We can find the
-	// change with serverSeq and documentID in the server. If the change is not
+	// change with serverSeq, documentKey and documentID in the server. If the change is not
 	// stored on the server, serverSeq is 0.
 	serverSeq int64
 

server/backend/database/doc_info.go

@@ -40,7 +40,7 @@ type DocInfo struct {
 	// OwnerKey is the key of the document owner.
 	OwnerKey string `bson:"owner_key"`
 
-	// OwnerKey is the ID of the document owner.
+	// OwnerID is the ID of the document owner.
 	OwnerID types.ID `bson:"owner_id"`
 
 	// CreatedAt is the time when the document is created.

server/backend/sync/coordinator.go

@@ -22,7 +22,6 @@ import (
 	gotime "time"
 
 	"github.com/yorkie-team/yorkie/api/types"
-	"github.com/yorkie-team/yorkie/pkg/document/key"
 	"github.com/yorkie-team/yorkie/pkg/document/time"
 )
 
@@ -42,15 +41,13 @@ type Coordinator interface {
 	Subscribe(
 		ctx context.Context,
 		subscriber *time.ActorID,
-		documentKey key.Key,
-		documentID types.ID,
+		documentRef types.DocRefKey,
 	) (*Subscription, []*time.ActorID, error)
 
 	// Unsubscribe unsubscribes from the given documents.
 	Unsubscribe(
 		ctx context.Context,
-		documentKey key.Key,
-		documentID types.ID,
+		documentRef types.DocRefKey,
 		sub *Subscription,
 	) error
 

server/backend/sync/memory/coordinator.go

@@ -21,7 +21,6 @@ import (
 	"context"
 
 	"github.com/yorkie-team/yorkie/api/types"
-	"github.com/yorkie-team/yorkie/pkg/document/key"
 	"github.com/yorkie-team/yorkie/pkg/document/time"
 	"github.com/yorkie-team/yorkie/pkg/locker"
 	"github.com/yorkie-team/yorkie/server/backend/sync"
@@ -59,26 +58,24 @@ func (c *Coordinator) NewLocker(
 func (c *Coordinator) Subscribe(
 	ctx context.Context,
 	subscriber *time.ActorID,
-	documentKey key.Key,
-	documentID types.ID,
+	documentRef types.DocRefKey,
 ) (*sync.Subscription, []*time.ActorID, error) {
-	sub, err := c.pubSub.Subscribe(ctx, subscriber, documentKey, documentID)
+	sub, err := c.pubSub.Subscribe(ctx, subscriber, documentRef)
 	if err != nil {
 		return nil, nil, err
 	}
 
-	ids := c.pubSub.ClientIDs(documentKey, documentID)
+	ids := c.pubSub.ClientIDs(documentRef)
 	return sub, ids, nil
 }
 
 // Unsubscribe unsubscribes the given documents.
 func (c *Coordinator) Unsubscribe(
 	ctx context.Context,
-	documentKey key.Key,
-	documentID types.ID,
+	documentRef types.DocRefKey,
 	sub *sync.Subscription,
 ) error {
-	c.pubSub.Unsubscribe(ctx, documentKey, documentID, sub)
+	c.pubSub.Unsubscribe(ctx, documentRef, sub)
 	return nil
 }
 

server/backend/sync/memory/coordinator_test.go

@@ -33,13 +33,17 @@ func TestCoordinator(t *testing.T) {
 		coordinator := memory.NewCoordinator(nil)
 		docKey := key.Key(t.Name() + "key")
 		docID := types.ID(t.Name() + "id")
+		docRef := types.DocRefKey{
+			Key: docKey,
+			ID:  docID,
+		}
 		ctx := context.Background()
 
 		for i := 0; i < 5; i++ {
 			id, err := time.ActorIDFromBytes([]byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, byte(i)})
 			assert.NoError(t, err)
 
-			_, clientIDs, err := coordinator.Subscribe(ctx, id, docKey, docID)
+			_, clientIDs, err := coordinator.Subscribe(ctx, id, docRef)
 			assert.NoError(t, err)
 			assert.Len(t, clientIDs, i+1)
 		}

server/backend/sync/memory/pubsub.go

@@ -24,7 +24,6 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/yorkie-team/yorkie/api/types"
-	"github.com/yorkie-team/yorkie/pkg/document/key"
 	"github.com/yorkie-team/yorkie/pkg/document/time"
 	"github.com/yorkie-team/yorkie/server/backend/sync"
 	"github.com/yorkie-team/yorkie/server/logging"
@@ -67,28 +66,27 @@ func (s *subscriptions) Len() int {
 // PubSub is the memory implementation of PubSub, used for single server.
 type PubSub struct {
 	subscriptionsMapMu    *gosync.RWMutex
-	subscriptionsMapByDoc map[key.Key]map[types.ID]*subscriptions
+	subscriptionsMapByDoc map[types.DocRefKey]*subscriptions
 }
 
 // NewPubSub creates an instance of PubSub.
 func NewPubSub() *PubSub {
 	return &PubSub{
 		subscriptionsMapMu:    &gosync.RWMutex{},
-		subscriptionsMapByDoc: make(map[key.Key]map[types.ID]*subscriptions),
+		subscriptionsMapByDoc: make(map[types.DocRefKey]*subscriptions),
 	}
 }
 
 // Subscribe subscribes to the given document keys.
 func (m *PubSub) Subscribe(
 	ctx context.Context,
 	subscriber *time.ActorID,
-	documentKey key.Key,
-	documentID types.ID,
+	documentRef types.DocRefKey,
 ) (*sync.Subscription, error) {
 	if logging.Enabled(zap.DebugLevel) {
 		logging.From(ctx).Debugf(
-			`Subscribe(%s.%s,%s) Start`,
-			documentKey, documentID,
+			`Subscribe(%s,%s) Start`,
+			documentRef,
 			subscriber.String(),
 		)
 	}
@@ -97,18 +95,15 @@ func (m *PubSub) Subscribe(
 	defer m.subscriptionsMapMu.Unlock()
 
 	sub := sync.NewSubscription(subscriber)
-	if _, ok := m.subscriptionsMapByDoc[documentKey]; !ok {
-		m.subscriptionsMapByDoc[documentKey] = make(map[types.ID]*subscriptions)
+	if _, ok := m.subscriptionsMapByDoc[documentRef]; !ok {
+		m.subscriptionsMapByDoc[documentRef] = newSubscriptions()
 	}
-	if _, ok := m.subscriptionsMapByDoc[documentKey][documentID]; !ok {
-		m.subscriptionsMapByDoc[documentKey][documentID] = newSubscriptions()
-	}
-	m.subscriptionsMapByDoc[documentKey][documentID].Add(sub)
+	m.subscriptionsMapByDoc[documentRef].Add(sub)
 
 	if logging.Enabled(zap.DebugLevel) {
 		logging.From(ctx).Debugf(
-			`Subscribe(%s.%s,%s) End`,
-			documentKey, documentID,
+			`Subscribe(%s,%s) End`,
+			documentRef,
 			subscriber.String(),
 		)
 	}
@@ -118,40 +113,34 @@ func (m *PubSub) Subscribe(
 // Unsubscribe unsubscribes the given docKeys.
 func (m *PubSub) Unsubscribe(
 	ctx context.Context,
-	documentKey key.Key,
-	documentID types.ID,
+	documentRef types.DocRefKey,
 	sub *sync.Subscription,
 ) {
 	m.subscriptionsMapMu.Lock()
 	defer m.subscriptionsMapMu.Unlock()
 
 	if logging.Enabled(zap.DebugLevel) {
 		logging.From(ctx).Debugf(
-			`Unsubscribe(%s.%s,%s) Start`,
-			documentKey, documentID,
+			`Unsubscribe(%s,%s) Start`,
+			documentRef,
 			sub.Subscriber().String(),
 		)
 	}
 
 	sub.Close()
 
-	if subsByDocID, ok := m.subscriptionsMapByDoc[documentKey]; ok {
-		if subs, ok := subsByDocID[documentID]; ok {
-			subs.Delete(sub.ID())
+	if subs, ok := m.subscriptionsMapByDoc[documentRef]; ok {
+		subs.Delete(sub.ID())
 
-			if subs.Len() == 0 {
-				delete(m.subscriptionsMapByDoc[documentKey], documentID)
-			}
-		}
-		if len(subsByDocID) == 0 {
-			delete(m.subscriptionsMapByDoc, documentKey)
+		if subs.Len() == 0 {
+			delete(m.subscriptionsMapByDoc, documentRef)
 		}
 	}
 
 	if logging.Enabled(zap.DebugLevel) {
 		logging.From(ctx).Debugf(
-			`Unsubscribe(%s.%s,%s) End`,
-			documentKey, documentID,
+			`Unsubscribe(%s,%s) End`,
+			documentRef,
 			sub.Subscriber().String(),
 		)
 	}
@@ -166,60 +155,61 @@ func (m *PubSub) Publish(
 	m.subscriptionsMapMu.RLock()
 	defer m.subscriptionsMapMu.RUnlock()
 
-	documentKey := event.DocumentKey
-	documentID := event.DocumentID
+	documentRef := event.DocumentRef
 	if logging.Enabled(zap.DebugLevel) {
-		logging.From(ctx).Debugf(`Publish(%s.%s,%s) Start`,
-			documentKey, documentID,
-			publisherID.String())
+		logging.From(ctx).Debugf(
+			`Publish(%s,%s) Start`,
+			documentRef,
+			publisherID.String(),
+		)
 	}
 
-	if subsByDocID, ok := m.subscriptionsMapByDoc[documentKey]; ok {
-		if subs, ok := subsByDocID[documentID]; ok {
-			for _, sub := range subs.Map() {
-				if sub.Subscriber().Compare(publisherID) == 0 {
-					continue
-				}
-
-				if logging.Enabled(zap.DebugLevel) {
-					logging.From(ctx).Debugf(
-						`Publish %s(%s.%s,%s) to %s`,
-						event.Type,
-						documentKey, documentID,
-						publisherID.String(),
-						sub.Subscriber().String(),
-					)
-				}
-
-				// NOTE: When a subscription is being closed by a subscriber,
-				// the subscriber may not receive messages.
-				select {
-				case sub.Events() <- event:
-				case <-gotime.After(100 * gotime.Millisecond):
-					logging.From(ctx).Warnf(
-						`Publish(%s.%s,%s) to %s timeout`,
-						documentKey, documentID,
-						publisherID.String(),
-						sub.Subscriber().String(),
-					)
-				}
+	if subs, ok := m.subscriptionsMapByDoc[documentRef]; ok {
+		for _, sub := range subs.Map() {
+			if sub.Subscriber().Compare(publisherID) == 0 {
+				continue
+			}
+
+			if logging.Enabled(zap.DebugLevel) {
+				logging.From(ctx).Debugf(
+					`Publish %s(%s,%s) to %s`,
+					event.Type,
+					documentRef,
+					publisherID.String(),
+					sub.Subscriber().String(),
+				)
+			}
+
+			// NOTE: When a subscription is being closed by a subscriber,
+			// the subscriber may not receive messages.
+			select {
+			case sub.Events() <- event:
+			case <-gotime.After(100 * gotime.Millisecond):
+				logging.From(ctx).Warnf(
+					`Publish(%s,%s) to %s timeout`,
+					documentRef,
+					publisherID.String(),
+					sub.Subscriber().String(),
+				)
 			}
 		}
 	}
+
 	if logging.Enabled(zap.DebugLevel) {
-		logging.From(ctx).Debugf(`Publish(%s.%s,%s) End`,
-			documentKey, documentID,
+		logging.From(ctx).Debugf(
+			`Publish(%s,%s) End`,
+			documentRef,
 			publisherID.String())
 	}
 }
 
 // ClientIDs returns the clients of the given document.
-func (m *PubSub) ClientIDs(documentKey key.Key, documentID types.ID) []*time.ActorID {
+func (m *PubSub) ClientIDs(documentRef types.DocRefKey) []*time.ActorID {
 	m.subscriptionsMapMu.RLock()
 	defer m.subscriptionsMapMu.RUnlock()
 
 	var ids []*time.ActorID
-	for _, sub := range m.subscriptionsMapByDoc[documentKey][documentID].Map() {
+	for _, sub := range m.subscriptionsMapByDoc[documentRef].Map() {
 		ids = append(ids, sub.Subscriber())
 	}
 	return ids

server/backend/sync/memory/pubsub_test.go

@@ -40,19 +40,22 @@ func TestPubSub(t *testing.T) {
 		pubSub := memory.NewPubSub()
 		key := key.Key(t.Name() + "key")
 		id := types.ID(t.Name() + "id")
+		docRef := types.DocRefKey{
+			Key: key,
+			ID:  id,
+		}
 		docEvent := sync.DocEvent{
 			Type:        types.DocumentWatchedEvent,
 			Publisher:   idB,
-			DocumentKey: key,
-			DocumentID:  id,
+			DocumentRef: docRef,
 		}
 
 		ctx := context.Background()
 		// subscribe the documents by actorA
-		subA, err := pubSub.Subscribe(ctx, idA, key, id)
+		subA, err := pubSub.Subscribe(ctx, idA, docRef)
 		assert.NoError(t, err)
 		defer func() {
-			pubSub.Unsubscribe(ctx, key, id, subA)
+			pubSub.Unsubscribe(ctx, docRef, subA)
 		}()
 
 		var wg gosync.WaitGroup

server/backend/sync/pubsub.go

@@ -20,7 +20,6 @@ import (
 	"github.com/rs/xid"
 
 	"github.com/yorkie-team/yorkie/api/types"
-	"github.com/yorkie-team/yorkie/pkg/document/key"
 	"github.com/yorkie-team/yorkie/pkg/document/time"
 )
 
@@ -50,8 +49,7 @@ func (s *Subscription) ID() string {
 type DocEvent struct {
 	Type        types.DocEventType
 	Publisher   *time.ActorID
-	DocumentKey key.Key
-	DocumentID  types.ID
+	DocumentRef types.DocRefKey
 	Body        types.DocEventBody
 }