package processor

import (
	"context"
	"encoding/json"
	"os"
	"sync"
	"testing"
	"time"

	"github.com/DoNewsCode/core"
	"github.com/DoNewsCode/core/di"
	"github.com/DoNewsCode/core/otkafka"
	"github.com/pkg/errors"
	"github.com/segmentio/kafka-go"
	"github.com/stretchr/testify/assert"
)

type testData struct {
	ID int `json:"id"`
}

type testHandlerA struct {
	data chan *testData
}

func (h *testHandlerA) Info() *Info {
	return &Info{
		Name:        "A",
		BatchSize:   3,
		ReadWorker:  2,
		BatchWorker: 2,
	}
}

func (h *testHandlerA) Handle(_ context.Context, msg *kafka.Message) (interface{}, error) {
	e := &testData{}
	if err := json.Unmarshal(msg.Value, &e); err != nil {
		return nil, err
	}
	return e, nil
}

func (h *testHandlerA) Batch(_ context.Context, data []interface{}) error {
	for _, e := range data {
		h.data