// Copyright 2020 the go-etl Authors.
//
// 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 file

import (
	"context"

	"github.com/Breeze0806/go-etl/config"
	"github.com/Breeze0806/go-etl/datax/common/plugin"
	"github.com/Breeze0806/go-etl/element"
	"github.com/Breeze0806/go-etl/storage/stream/file"
)

// Task  normal file task
type Task struct {
	*plugin.BaseTask

	streamer *file.InStreamer
}

// NewTask Create normal file task
func NewTask() *Task {
	return &Task{
		BaseTask: plugin.NewBaseTask(),
	}
}

// Init Initialize the task
func (t *Task) Init(ctx context.Context) (err error) {
	var name string
	if name, err = t.PluginConf().GetString("opener"); err != nil {
		return t.Wrapf(err, "GetString fail")
	}
	var filename string
	if filename, err = t.PluginJobConf().GetString("path"); err != nil {
		return t.Wrapf(err, "GetString fail")
	}

	if t.streamer, err = file.NewInStreamer(name, filename); err != nil {
		return t.Wrapf(err, "GetString fail")
	}
	return
}

// Destroy Destroy normal file task
func (t *Task) Destroy(ctx context.Context) (err error) {
	if t.streamer != nil {
		err = t.streamer.Close()
	}
	return t.Wrapf(err, "Close fail")
}

type handler struct {
	sender plugin.RecordSender
}

func newHander(sender plugin.RecordSender) *handler {
	return &handler{
		sender: sender,
	}
}

func (h *handler) CreateRecord() (element.Record, error) {
	return h.sender.CreateRecord()
}

func (h *handler) OnRecord(r element.Record) error {
	return h.sender.SendWriter(r)
}

// StartRead Start reading data and send it to the sender
func (t *Task) StartRead(ctx context.Context, sender plugin.RecordSender) (err error) {
	handler := newHander(sender)

	log.Infof(t.Format("startRead begin"))
	defer func() {
		sender.Terminate()
		log.Infof(t.Format("startRead end"))
	}()
	var configs []*config.JSON
	configs, err = t.PluginJobConf().GetConfigArray("content")
	if err != nil {
		return t.Wrapf(err, "GetConfigArray fail.")
	}

	for _, conf := range configs {
		if err = t.streamer.Read(ctx, conf, handler); err != nil {
			return t.Wrapf(err, "Read fail.")
		}
	}
	return nil
}
