fix: update messages type to topic

tigrisdata-archive · Oct 1, 2022 · 0e8dfbe · 0e8dfbe
1 parent e174ab1
commit 0e8dfbe
Show file tree

Hide file tree

Showing 5 changed files with 10 additions and 10 deletions.
diff --git a/schema/collection.go b/schema/collection.go
@@ -68,7 +68,7 @@ type CollectionType string
 
 const (
 	DocumentsType CollectionType = "documents"
-	MessagesType  CollectionType = "messages"
+	TopicType     CollectionType = "topic"
 )
 
 func disableAdditionalProperties(properties map[string]*jsonschema.Schema) {

diff --git a/schema/schema.go b/schema/schema.go
@@ -107,8 +107,8 @@ func GetCollectionType(reqSchema jsoniter.RawMessage) (CollectionType, error) {
 		switch string(val) {
 		case "documents":
 			return DocumentsType, nil
-		case "messages":
-			return MessagesType, nil
+		case "messages", "topic":
+			return TopicType, nil
 		}
 	}
 	if dt == jsonparser.NotExist {
@@ -125,7 +125,7 @@ func Build(collection string, reqSchema jsoniter.RawMessage) (*Factory, error) {
 		return nil, err
 	}
 
-	if cType != MessagesType {
+	if cType != TopicType {
 		if reqSchema, err = setPrimaryKey(reqSchema, jsonSpecFormatUUID, true); err != nil {
 			return nil, err
 		}
@@ -144,7 +144,7 @@ func Build(collection string, reqSchema jsoniter.RawMessage) (*Factory, error) {
 
 	if len(schema.PrimaryKeys) == 0 && cType == DocumentsType {
 		return nil, errors.InvalidArgument("missing primary key field in schema")
-	} else if len(schema.PrimaryKeys) > 0 && cType == MessagesType {
+	} else if len(schema.PrimaryKeys) > 0 && cType == TopicType {
 		return nil, errors.InvalidArgument("setting primary key is not supported for messages collection")
 	}
 

diff --git a/schema/schema_test.go b/schema/schema_test.go
@@ -366,10 +366,10 @@ func TestGetCollectionType(t *testing.T) {
 			"type": "integer"
 		}
 	},
-    "collection_type": "messages"
+    "collection_type": "topic"
 }`)
 
 	ty, err = GetCollectionType(schema)
-	require.Equal(t, MessagesType, ty)
+	require.Equal(t, TopicType, ty)
 	require.NoError(t, err)
 }
diff --git a/server/services/v1/query_runner.go b/server/services/v1/query_runner.go
@@ -325,7 +325,7 @@ func (runner *BaseQueryRunner) mustBeDocumentsCollection(collection *schema.Defa
 }
 
 func (runner *BaseQueryRunner) mustBeMessagesCollection(collection *schema.DefaultCollection, method string) error {
-	if collection.Type() != schema.MessagesType {
+	if collection.Type() != schema.TopicType {
 		return errors.InvalidArgument("%s is only supported on collection type of 'messages'", method)
 	}
 
@@ -652,7 +652,7 @@ func (runner *StreamingQueryRunner) buildReaderOptions(tenant *metadata.Tenant,
 		}
 	}
 
-	if collection.Type() == schema.MessagesType {
+	if collection.Type() == schema.TopicType {
 		// if it is event streaming then fallback to indexing store for all reads
 		if !config.IsIndexingStoreReadEnabled() {
 			if options.from == nil {

diff --git a/test/v1/client/driver_test.go b/test/v1/client/driver_test.go
@@ -488,7 +488,7 @@ func testPubSub(t *testing.T, c driver.Driver) {
 				"type": "boolean"
 			}
 		},
-        "collection_type": "messages"
+        "collection_type": "topic"
 	}`
 
 	err := c.CreateDatabase(ctx, dbName)