wip

cmd/verify/confirm/confirm.go

@@ -3,13 +3,14 @@ package confirm
 import (
 	"context"
 	"fmt"
-	"github.com/segmentio/kafka-go"
-	"github.com/spf13/cobra"
 	"hermetic/internal/common_flags"
 	"hermetic/internal/teams"
 	confirmImplementation "hermetic/internal/verify/confirm"
 	"os"
 	"os/signal"
+
+	"github.com/segmentio/kafka-go"
+	"github.com/spf13/cobra"
 )
 
 func NewCommand() *cobra.Command {

internal/dps/read.go

@@ -4,40 +4,34 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+
 	"github.com/segmentio/kafka-go"
 )
 
-func ReadMessages(ctx context.Context, reader *kafka.Reader, callback func(*KafkaResponse) error) error {
+func ReadMessage(ctx context.Context, reader *kafka.Reader) (*KafkaResponse, error) {
 	for {
-		// TODO log debug
 		fmt.Println("Reading next message...")
 		message, err := reader.ReadMessage(ctx)
 		if err != nil {
-			return fmt.Errorf("failed to read message: %w", err)
+			return nil, fmt.Errorf("failed to read message: %w", err)
 		}
 
 		var dpsResponse DigitalPreservationSystemResponse
 
 		err = json.Unmarshal(message.Value, &dpsResponse)
 		if err != nil {
-			return fmt.Errorf("could not unmarshal message at offset '%d': %w", message.Offset, err)
+			return nil, fmt.Errorf("could not unmarshal message at offset '%d': %w", message.Offset, err)
 		}
 
 		if !IsWebArchiveOwned(&dpsResponse) {
-			// TODO log debug message
 			fmt.Printf("Message at offset '%d' is not owned by web archive, skipping offset\n", message.Offset)
 			continue
 		}
 
-		response := &KafkaResponse{
+		return &KafkaResponse{
 			Offset:      message.Offset,
 			Key:         string(message.Key),
 			DPSResponse: dpsResponse,
-		}
-
-		err = callback(response)
-		if err != nil {
-			return err
-		}
+		}, nil
 	}
 }

internal/verify/confirm/confirm.go

@@ -3,24 +3,18 @@ package confirmImplmentation
 import (
 	"context"
 	"fmt"
-	"github.com/segmentio/kafka-go"
 	"hermetic/internal/dps"
+
+	"github.com/segmentio/kafka-go"
 )
 
 func ReadConfirmTopic(ctx context.Context, reader *kafka.Reader) error {
-	err := dps.ReadMessages(ctx, reader, ProcessMessagesFromConfirmTopic)
-	if err != nil {
-		return fmt.Errorf("failed to read confirm-topic: `%w`", err)
-	}
-	return nil
-}
-
-func ProcessMessagesFromConfirmTopic(response *dps.KafkaResponse) error {
-	if response == nil {
-		panic("No response found")
+	for {
+		response, err := dps.ReadMessage(ctx, reader)
+		if err != nil {
+			return fmt.Errorf("failed to read confirm-topic: `%w`", err)
+		}
+
+		fmt.Printf("SIP successfully preserved: %+v\n", response)
 	}
-
-	fmt.Printf("SIP successfully preserved: %+v\n", response)
-
-	return nil
 }

internal/verify/reject/reject.go

@@ -3,30 +3,27 @@ package rejectImplementation
 import (
 	"context"
 	"fmt"
-	"github.com/segmentio/kafka-go"
 	"hermetic/internal/dps"
 	"hermetic/internal/teams"
+
+	"github.com/segmentio/kafka-go"
 )
 
 func ReadRejectTopic(ctx context.Context, reader *kafka.Reader, teamsWebhookNotificationUrl string) error {
-	err := dps.ReadMessages(ctx, reader, func(response *dps.KafkaResponse) error {
-		if err := ProcessMessagesFromRejectTopic(reader, response, teamsWebhookNotificationUrl); err != nil {
-			return fmt.Errorf("failed to process message from reject-topic: `%w`", err)
+	for {
+		response, err := dps.ReadMessage(ctx, reader)
+		if err != nil {
+			return fmt.Errorf("failed to read reject-topic: `%w`", err)
 		}
-		return nil
-	})
-	if err != nil {
-		return fmt.Errorf("failed to read reject-topic: `%w`", err)
+
+		if err := ProcessMessagesFromRejectTopic(reader, response, teamsWebhookNotificationUrl); err != nil {
+				return fmt.Errorf("failed to process message from reject-topic: `%w`", err)
+		}	
 	}
-	return nil
 }
 
 func ProcessMessagesFromRejectTopic(reader *kafka.Reader, response *dps.KafkaResponse, teamsWebhookNotificationUrl string) error {
-	if response == nil {
-		panic("No response found")
-	}
-
-	// TODO log more info from reject message in case sending to teams fails
+	// TODO print more info from reject message in case sending to teams fails
 	fmt.Printf("Processing message with ContentCategory: '%s'\n", response.DPSResponse.ContentCategory)
 
 	payload := createTeamsDigitalPreservationSystemFailureMessage(response, reader.Config().Topic, reader.Config().Brokers)