package main

import (
	"context"
	"fmt"
	"log"
	"math/rand"
	"sync/atomic"
	"time"

	"github.com/Shopify/sarama"

	"github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2"
	cloudevents "github.com/cloudevents/sdk-go/v2"
)

const (
	count = 10
)

func main() {
	saramaConfig := sarama.NewConfig()
	saramaConfig.Version = sarama.V2_0_0_0

	topic := fmt.Sprintf("topic-%v", time.Now().Unix())

	receiver, err := kafka_sarama.NewConsumer([]string{"127.0.0.1:9092"}, saramaConfig, "test-group-id", topic)
	if err != nil {
		log.Fatalf("failed to create receiver: %s", err.Error())
	}

	defer receiver.Close(context.Background())

	c, err := cloudevents.NewClient(receiver, cloudevents.WithPollGoroutines(1))
	if err != nil {
		log.Fatalf("failed to create client, %v", err)
	}

	// Create a done channel to block until we've received (count) messages
	done := make(chan struct{})

	// Start the receiver
	go func() {
		log.Printf("will listen consuming topic %v\n", topic)
		var recvCount int32
		err = c.StartReceiver(context.TODO(), func(ctx context.Context, event cloudevents.Event) {
			receive(ctx, event)
			if atomic.AddInt32(&recvCount, 1) == count {
				done <- struct{}{}
			}
		})
		if err != nil {
			log.Fatalf("failed to start receiver: %s", err)
		} else {
			log.Printf("receiver stopped\n")
		}
	}()

	// Ensure that the consumer is ready before staring pushing events
	time.Sleep(4 * time.Second)

	sender, err := kafka_sarama.NewSender([]string{"127.0.0.1:9092"}, saramaConfig, topic)
	if err != nil {
		log.Fatalf("failed to create protocol: %s", err.Error())
	}

	defer sender.Close(context.Background())

	cp, err := cloudevents.NewClient(sender, cloudevents.WithTimeNow(), cloudevents.WithUUIDs(), cloudevents.WithPollGoroutines(1))
	if err != nil {
		log.Fatalf("failed to create client, %v", err)
	}

	// Start sending the events
	for i := 0; i < count; i++ {
		e := cloudevents.NewEvent()
		e.SetType("com.cloudevents.sample.sent")
		e.SetSource("https://github.com/cloudevents/sdk-go/v2/samples/httpb/requester")
		_ = e.SetData(cloudevents.ApplicationJSON, map[string]interface{}{
			"id":      i,
			"message": "Hello, World!",
		})

		if result := cp.Send(context.Background(), e); cloudevents.IsUndelivered(result) {
			log.Printf("failed to send: %v", err)
		} else {
			log.Printf("sent: %d, accepted: %t", i, cloudevents.IsACK(result))
		}
	}

	<-done
}

func receive(ctx context.Context, event cloudevents.Event) {
	// Random processing time
	n := rand.Intn(10)
	time.Sleep(time.Duration(n) * time.Second)
	
	fmt.Printf("After %vs, %s\n", n, event.DataEncoded)
}

	wg := sync.WaitGroup{}

	type job struct {
		Msg    binding.Message
		RespFn protocol.ResponseFn
	}

	// chan to hold message to process.
	jobs := make(chan job)

	// Start N processors.
	for i := 0; i < c.processGoroutines; i++ {
		go func() {
			for job := range jobs {
				wg.Add(1)
				defer wg.Done()

				if err := c.invoker.Invoke(ctx, job.Msg, job.RespFn); err != nil {
					cecontext.LoggerFrom(ctx).Warnf("Error while handling a message: %s", err)
				}
			}

			cecontext.LoggerFrom(ctx).Warn("Processor stop working")
			// Restart it ?
		}()
	}

	// Start Polling.
	for i := 0; i < c.pollGoroutines; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				var msg binding.Message
				var respFn protocol.ResponseFn
				var err error

				if c.responder != nil {
					msg, respFn, err = c.responder.Respond(ctx)
				} else if c.receiver != nil {
					msg, err = c.receiver.Receive(ctx)
					respFn = noRespFn
				}

				if err == io.EOF { // Normal close
					return
				}

				if err != nil {
					cecontext.LoggerFrom(ctx).Warnf("Error while receiving a message: %s", err)
					continue
				}

				// Add job to queue
				jobs <- job{Msg: msg, RespFn: respFn}
			}
		}()
	}