[Schemas] Connector SDK: Create and get a schema from Conduit

Makefile

@@ -16,7 +16,7 @@ proto-generate:
 
 .PHONY: proto-update
 proto-update:
-	cd proto && buf mod update
+	cd proto && buf dep update
 
 .PHONY: proto-lint
 proto-lint:

conduit/pschema/errors.go

@@ -0,0 +1,21 @@
+// Copyright © 2024 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package pschema
+
+import "errors"
+
+var (
+	ErrSchemaNotFound = errors.New("schema not found")
+)

conduit/pschema/in_memory.go

@@ -0,0 +1,68 @@
+// Copyright © 2024 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package pschema
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/conduitio/conduit-commons/schema"
+)
+
+type inMemoryService struct {
+	// schemas is a map of schema subjects to all the versions of that schema
+	// versioning starts at 1, newer versions are appended to the end of the versions slice.
+	schemas map[string][]schema.Instance
+	// m guards access to schemas
+	m sync.Mutex
+}
+
+func NewInMemoryService() Service {
+	return &inMemoryService{
+		schemas: make(map[string][]schema.Instance),
+	}
+}
+
+func (s *inMemoryService) Create(_ context.Context, request CreateRequest) (CreateResponse, error) {
+	s.m.Lock()
+	defer s.m.Unlock()
+
+	inst := schema.Instance{
+		Subject: request.Subject,
+		Version: len(s.schemas[request.Subject]) + 1,
+		Type:    schema.Type(request.Type),
+		Bytes:   request.Bytes,
+	}
+	s.schemas[request.Subject] = append(s.schemas[request.Subject], inst)
+
+	return CreateResponse{Instance: inst}, nil
+}
+
+func (s *inMemoryService) Get(_ context.Context, request GetRequest) (GetResponse, error) {
+	s.m.Lock()
+	defer s.m.Unlock()
+
+	versions, ok := s.schemas[request.Subject]
+	if !ok {
+		return GetResponse{}, fmt.Errorf("name %v: %w", request.Subject, ErrSchemaNotFound)
+	}
+
+	if len(versions) < request.Version {
+		return GetResponse{}, fmt.Errorf("version %v: %w", request.Version, ErrSchemaNotFound)
+	}
+
+	return GetResponse{Instance: versions[request.Version-1]}, nil
+}

conduit/pschema/in_memory_test.go

@@ -0,0 +1,72 @@
+// Copyright © 2024 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package pschema
+
+import (
+	"context"
+	"testing"
+
+	"github.com/conduitio/conduit-commons/schema"
+	"github.com/google/go-cmp/cmp"
+	"github.com/matryer/is"
+)
+
+func TestInMemoryService(t *testing.T) {
+	is := is.New(t)
+	ctx := context.Background()
+
+	underTest := NewInMemoryService()
+
+	want1 := schema.Instance{
+		Subject: "test-subject",
+		Version: 1,
+		Type:    schema.TypeAvro,
+		Bytes:   []byte("irrelevant 1"),
+	}
+
+	// Create first version
+	got1, err := underTest.Create(ctx, CreateRequest{
+		Subject: want1.Subject,
+		Type:    Type(want1.Type),
+		Bytes:   want1.Bytes,
+	})
+	is.NoErr(err)
+	is.Equal("", cmp.Diff(want1, got1.Instance))
+
+	// Create second version
+	want2 := schema.Instance{
+		Subject: want1.Subject,
+		Version: 2,
+		Type:    want1.Type,
+		Bytes:   []byte("irrelevant 2"),
+	}
+	got2, err := underTest.Create(ctx, CreateRequest{
+		Subject: want2.Subject,
+		Type:    Type(want2.Type),
+		Bytes:   want2.Bytes,
+	})
+	is.NoErr(err)
+	is.Equal("", cmp.Diff(want2, got2.Instance))
+
+	// Get first version
+	getResp1, err := underTest.Get(ctx, GetRequest{Subject: want1.Subject, Version: 1})
+	is.NoErr(err)
+	is.Equal("", cmp.Diff(want1, getResp1.Instance))
+
+	// Get second version
+	getResp2, err := underTest.Get(ctx, GetRequest{Subject: want2.Subject, Version: 2})
+	is.NoErr(err)
+	is.Equal("", cmp.Diff(want2, getResp2.Instance))
+}

conduit/schema/schema.go → conduit/pschema/schema.go

@@ -12,7 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package schema
+//go:generate mockgen -typed -destination=mock/pschema.go -package=mock -mock_names=Service=Service . Service
+
+package pschema
 
 import (
 	"context"
@@ -32,16 +34,17 @@ const (
 )
 
 type CreateRequest struct {
-	Name  string
-	Type  Type
-	Bytes []byte
+	Subject string
+	Type    Type
+	Bytes   []byte
 }
 type CreateResponse struct {
 	schema.Instance
 }
 
 type GetRequest struct {
-	ID string
+	Subject string
+	Version int
 }
 type GetResponse struct {
 	schema.Instance

conduit/schema/v1/fromproto/schema.go → conduit/pschema/v1/fromproto/schema.go

@@ -16,7 +16,7 @@ package fromproto
 
 import (
 	cschema "github.com/conduitio/conduit-commons/schema"
-	"github.com/conduitio/conduit-connector-protocol/conduit/schema"
+	"github.com/conduitio/conduit-connector-protocol/conduit/pschema"
 	conduitv1 "github.com/conduitio/conduit-connector-protocol/proto/conduit/v1"
 )
 
@@ -27,23 +27,33 @@ func _() {
 	_ = cTypes[int(cschema.TypeAvro)-int(conduitv1.Schema_TYPE_AVRO)]
 
 	// Compatibility between the protocol's schema.Type and the Protobuf schema type
-	_ = cTypes[int(schema.TypeAvro)-int(conduitv1.Schema_TYPE_AVRO)]
+	_ = cTypes[int(pschema.TypeAvro)-int(conduitv1.Schema_TYPE_AVRO)]
 }
 
-func CreateRequest(req *conduitv1.CreateSchemaRequest) schema.CreateRequest {
-	return schema.CreateRequest{
-		Name:  req.Name,
-		Type:  schema.Type(req.Type),
-		Bytes: req.Bytes,
+func CreateRequest(req *conduitv1.CreateSchemaRequest) pschema.CreateRequest {
+	return pschema.CreateRequest{
+		Subject: req.Subject,
+		Type:    pschema.Type(req.Type),
+		Bytes:   req.Bytes,
 	}
 }
 
-func CreateResponse(resp *conduitv1.CreateSchemaResponse) schema.CreateResponse {
-	return schema.CreateResponse{
+func CreateResponse(resp *conduitv1.CreateSchemaResponse) pschema.CreateResponse {
+	return pschema.CreateResponse{
 		Instance: cschema.Instance{
-			ID:      resp.Schema.Id,
-			Name:    resp.Schema.Name,
-			Version: resp.Schema.Version,
+			Subject: resp.Schema.Subject,
+			Version: int(resp.Schema.Version),
+			Type:    cschema.Type(resp.Schema.Type),
+			Bytes:   resp.Schema.Bytes,
+		},
+	}
+}
+
+func GetResponse(resp *conduitv1.GetSchemaResponse) pschema.GetResponse {
+	return pschema.GetResponse{
+		Instance: cschema.Instance{
+			Subject: resp.Schema.Subject,
+			Version: int(resp.Schema.Version),
 			Type:    cschema.Type(resp.Schema.Type),
 			Bytes:   resp.Schema.Bytes,
 		},