package main

import (
	"fmt"
	"net"

	log "github.com/sirupsen/logrus"
)

// UDPConfig udp config
type UDPConfig struct {
	IP          string                 `json:"ip,omitempty"`
	Port        uint16                 `json:"port,omitempty"`
	AuxProperty map[string]interface{} `json:"auxProperty,omitempty"`
}

// UDPReceiver udp receiver handle
type UDPReceiver struct {
	config UDPConfig
	conn   *net.UDPConn
	ch     chan rdata
}

// UDPData udp data
type UDPData struct {
	properties map[string]interface{}
	data       []byte
}

// Data implements Source interface
func (d UDPData) Data() []byte {
	return d.data
}

// GetProperty implements Source interface
func (d UDPData) GetProperty(prop string) (interface{}, bool) {
	p, ok := d.properties[prop]
	return p, ok
}

// Open open receiver
func (d *UDPReceiver) Open() error {
	bind := fmt.Sprintf("%s:%d", d.config.IP, d.config.Port)
	s4001addr, err := net.ResolveUDPAddr("udp", bind)
	if err != nil {
		return err
	}
	d.conn, err = net.ListenUDP("udp", s4001addr)
	if err != nil {
		return err
	}
	log.WithFields(log.Fields{"config": d.config}).Info("udp listen open")
	// defer s4001.Close() // nolint
	return nil
}

// Start start the receiver
func (d *UDPReceiver) Start() error {
	log.Info("udp listen start")
	go func() {
		for {
			var buf [4096]byte
			num, peer, err := d.conn.ReadFromUDP(buf[0:])
			if err != nil {
				log.WithFields(log.Fields{"err": err}).Error("error receive data")
			} else {
				log.WithFields(log.Fields{"peer": peer, "bytes": num}).Info("received data")
			}
			props := map[string]interface{}{
				"peer":   peer,
				"config": d.config,
			}
			for k, v := range d.config.AuxProperty {
				props[k] = v
			}
			d.ch <- UDPData{
				properties: props,
				data:       buf[0:num],
			}
		}
	}()
	return nil
}

// NewUDPReceiver create new receiver instance
func NewUDPReceiver(params UDPConfig, ch chan rdata) (UDPReceiver, error) {
	d := UDPReceiver{
		config: params,
		ch:     ch,
	}
	return d, nil
}