func (s *Server) Init(options []*agent.Option) error {
	req := &agent.Request{Message: &agent.Request_Init{
		Init: &agent.InitRequest{
			Options: options,
			TaskID:  s.taskID,
			NodeID:  s.nodeID,
		},
	}}
	resp, err := s.doRequestResponse(req, s.initResponse)
	if err != nil {
		return err
	}
	init := resp.Message.(*agent.Response_Init).Init
	if !init.Success {
		return fmt.Errorf("failed to initialize processes %s", init.Error)
	}
	return nil
}
Initialize the process with a set of Options.
Calling Init is required even if you do not have any specific Options, just pass nil
package udf
import (
	"errors"
	"fmt"
	"io"
	"log"
	"sync"
	"time"
	"github.com/influxdata/kapacitor/edge"
	"github.com/influxdata/kapacitor/models"
	"github.com/influxdata/kapacitor/udf/agent"
)
var ErrServerStopped = errors.New("server already stopped")
type Server struct {
	abortCallback func()
	abortOnce     sync.Once
	killCallback func()
	inMsg chan edge.Message
	outMsg chan edge.Message
	stopped  bool
	stopping chan struct{}
	aborted  bool
	aborting chan struct{}
	err   error
	errMu sync.Mutex
	requests      chan *agent.Request
	requestsGroup sync.WaitGroup
	keepalive        chan int64
	keepaliveTimeout time.Duration
	taskID string
	nodeID string
	in  agent.ByteReadReader
	out io.WriteCloser
	ioGroup sync.WaitGroup
	mu     sync.Mutex
	logger *log.Logger
	responseBuf []byte
	infoResponse     chan *agent.Response
	initResponse     chan *agent.Response
	snapshotResponse chan *agent.Response
	restoreResponse  chan *agent.Response
	begin  *agent.BeginBatch
	points []edge.BatchPointMessage
}
func NewServer(
	taskID, nodeID string,
	in agent.ByteReadReader,
	out io.WriteCloser,
	l *log.Logger,
	timeout time.Duration,
	abortCallback func(),
	killCallback func(),
) *Server {
	s := &Server{
		taskID:           taskID,
		nodeID:           nodeID,
		in:               in,
		out:              out,
		logger:           l,
		requests:         make(chan *agent.Request),
		keepalive:        make(chan int64, 1),
		keepaliveTimeout: timeout,
		abortCallback:    abortCallback,
		killCallback:     killCallback,
		inMsg:            make(chan edge.Message),
		outMsg:           make(chan edge.Message),
		infoResponse:     make(chan *agent.Response, 1),
		initResponse:     make(chan *agent.Response, 1),
		snapshotResponse: make(chan *agent.Response, 1),
		restoreResponse:  make(chan *agent.Response, 1),
	}
	return s
}
func (s *Server) In() chan<- edge.Message {
	return s.inMsg
}
func (s *Server) Out() <-chan edge.Message {
	return s.outMsg
}
func (s *Server) setError(err error) {
	s.errMu.Lock()
	defer s.errMu.Unlock()
	if s.err == nil {
		s.err = err
	}
}
func (s *Server) Start() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.stopped = false
	s.stopping = make(chan struct{})
	s.aborted = false
	s.aborting = make(chan struct{})
	s.ioGroup.Add(1)
	go func() {
		err := s.writeData()
		if err != nil {
			s.setError(err)
			defer s.abort()
		}
		s.ioGroup.Done()
	}()
	s.ioGroup.Add(1)
	go func() {
		err := s.readData()
		if err != nil {
			s.setError(err)
			defer s.abort()
		}
		s.ioGroup.Done()
	}()
	s.requestsGroup.Add(2)
	go s.runKeepalive()
	go s.watchKeepalive()
	return nil
}
func (s *Server) Abort(err error) {
	s.setError(err)
	s.abort()
}
func (s *Server) abort() {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.aborted {
		return
	}
	s.aborted = true
	close(s.aborting)
	if s.abortCallback != nil {
		s.abortOnce.Do(s.abortCallback)
	}
	_ = s.stop()
}
func (s *Server) WaitIO() {
	s.ioGroup.Wait()
}
func (s *Server) Stop() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.stop()
}
func (s *Server) stop() error {
	if s.stopped {
		return s.err
	}
	s.stopped = true
	close(s.stopping)
	s.requestsGroup.Wait()
	close(s.requests)
	close(s.inMsg)
	s.ioGroup.Wait()
	s.errMu.Lock()
	defer s.errMu.Unlock()
	return s.err
}
type Info struct {
	Wants    agent.EdgeType
	Provides agent.EdgeType
	Options  map[string]*agent.OptionInfo
}
func (s *Server) Info() (Info, error) {
	info := Info{}
	req := &agent.Request{Message: &agent.Request_Info{
		Info: &agent.InfoRequest{},
	}}
	resp, err := s.doRequestResponse(req, s.infoResponse)
	if err != nil {
		return info, err
	}
	ri := resp.Message.(*agent.Response_Info).Info
	info.Options = ri.Options
	info.Wants = ri.Wants
	info.Provides = ri.Provides
	return info, nil
}