refactor

command.go

@@ -25,8 +25,7 @@ type Command interface {
 }
 
 type CommandResponse interface {
-	// ErrStr returns an error string associated with this response.
-	ErrStr() string
+	isCommandResponse()
 }
 
 type SpecifyCmd struct{}
@@ -41,12 +40,7 @@ type SpecifyResponse struct {
 	Err           error
 }
 
-func (r *SpecifyResponse) ErrStr() string {
-	if r.Err == nil {
-		return ""
-	}
-	return r.Err.Error()
-}
+func (r *SpecifyResponse) isCommandResponse() {}
 
 type ConfigureCmd struct {
 	ConfigMap map[string]string
@@ -62,12 +56,7 @@ type ConfigureResponse struct {
 	Err error
 }
 
-func (r *ConfigureResponse) ErrStr() string {
-	if r.Err == nil {
-		return ""
-	}
-	return r.Err.Error()
-}
+func (r *ConfigureResponse) isCommandResponse() {}
 
 type OpenCmd struct{}
 
@@ -79,12 +68,7 @@ type OpenResponse struct {
 	Err error
 }
 
-func (r *OpenResponse) ErrStr() string {
-	if r.Err == nil {
-		return ""
-	}
-	return r.Err.Error()
-}
+func (r *OpenResponse) isCommandResponse() {}
 
 type ProcessCmd struct {
 	Records []opencdc.Record
@@ -98,9 +82,7 @@ type ProcessResponse struct {
 	Records []ProcessedRecord
 }
 
-func (r *ProcessResponse) ErrStr() string {
-	return ""
-}
+func (r *ProcessResponse) isCommandResponse() {}
 
 type TeardownCmd struct{}
 
@@ -114,9 +96,4 @@ type TeardownResponse struct {
 	Err error
 }
 
-func (r *TeardownResponse) ErrStr() string {
-	if r.Err == nil {
-		return ""
-	}
-	return r.Err.Error()
-}
+func (r *TeardownResponse) isCommandResponse() {}

internal/proto/command_to_proto.go → internal/proto/to_proto.go

@@ -122,19 +122,19 @@ func MarshalCommandResponse(resp sdk.CommandResponse) ([]byte, error) {
 				Version:     v.Specification.Version,
 				Author:      v.Specification.Author,
 				Parameters:  protoSpecificationParams(v.Specification.Parameters),
-				Err:         v.ErrStr(),
+				Err:         errorToString(v.Err),
 			},
 		}
 	case *sdk.ConfigureResponse:
 		protoResp.Response = &procproto.CommandResponse_ConfigureResp{
 			ConfigureResp: &procproto.Configure_Response{
-				Err: v.ErrStr(),
+				Err: errorToString(v.Err),
 			},
 		}
 	case *sdk.OpenResponse:
 		protoResp.Response = &procproto.CommandResponse_OpenResp{
 			OpenResp: &procproto.Open_Response{
-				Err: v.ErrStr(),
+				Err: errorToString(v.Err),
 			},
 		}
 	case *sdk.ProcessResponse:
@@ -150,7 +150,7 @@ func MarshalCommandResponse(resp sdk.CommandResponse) ([]byte, error) {
 	case *sdk.TeardownResponse:
 		protoResp.Response = &procproto.CommandResponse_TeardownResp{
 			TeardownResp: &procproto.Teardown_Response{
-				Err: v.ErrStr(),
+				Err: errorToString(v.Err),
 			},
 		}
 	default:
@@ -165,6 +165,14 @@ func MarshalCommandResponse(resp sdk.CommandResponse) ([]byte, error) {
 	return bytes, nil
 }
 
+func errorToString(err error) string {
+	if err == nil {
+		return ""
+	}
+
+	return err.Error()
+}
+
 func protoSpecificationParams(in map[string]sdk.Parameter) map[string]*procproto.Specify_Parameter {
 	out := make(map[string]*procproto.Specify_Parameter, len(in))
 	for name, param := range in {

internal/wasm_commands.go → internal/wasm/command_actions.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package wasm
 
 import (
 	"fmt"

internal/wasm_imports.go → internal/wasm/imports.go

@@ -14,7 +14,7 @@
 
 //go:build wasm
 
-package internal
+package wasm
 
 // Imports `nextCommand` from the host, which retrieves
 // the next command for a processor.

internal/wasm_imports_stub.go → internal/wasm/imports_stub.go

@@ -13,13 +13,13 @@
 // limitations under the License.
 
 // The functions in this file are stubs of the functions defined
-// in wasm_imports.go.
+// in imports.go.
 // They exist to make it possible to test, lint
 // or generally run the code in a non-WASM environment.
 
 //go:build !wasm
 
-package internal
+package wasm
 
 func _nextCommand(_, _ uint32) uint32 {
 	panic("stub")

internal/wasm_memory.go → internal/wasm/memory.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package wasm
 
 import (
 	"fmt"

run/run.go

@@ -20,7 +20,7 @@ import (
 	"os"
 
 	sdk "github.com/conduitio/conduit-processor-sdk"
-	"github.com/conduitio/conduit-processor-sdk/internal"
+	"github.com/conduitio/conduit-processor-sdk/internal/wasm"
 )
 
 // Run is the 'entry point' for a processor. It runs a
@@ -30,14 +30,14 @@ import (
 // A processor plugin needs to call this function in its main() function.
 func Run(p sdk.Processor) {
 	for {
-		cmd, err := internal.NextCommand()
+		cmd, err := wasm.NextCommand()
 		if err != nil {
 			_, _ = fmt.Fprintf(os.Stderr, "failed retrieving next command: %v", err)
 			os.Exit(1)
 		}
 
 		resp := cmd.Execute(context.Background(), p)
-		err = internal.Reply(resp)
+		err = wasm.Reply(resp)
 		if err != nil {
 			_, _ = fmt.Fprintf(os.Stderr, "failed writing reply: %v\n", err)
 			os.Exit(1)