unnamed-project/unnamed-project.go

package main

//go:generate go-bindata -o assets/assets_gen.go -pkg assets public/...

import (
	"container/list"
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"net"
	"net/http"
	"os"
	"os/user"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"varia.zone/unnamed-project/assets"
)

const help = `unnamed-project: WIP

Options:
	-n    node identifier (default: your system username)
	-p    path to share files (default: current working directory)
	-w    port for web server (default: 1312)
	-s    port for file server (default: 5555)
	-d    debug logging
	-h    output help
`

var nodeIDFlag string
var pathFlag string
var webPortFlag int
var filePortFlag int
var debugFlag bool
var helpFlag bool

func main() {
	systemUser, err := user.Current()
	if err != nil {
		log.Fatalln(err)
	}

	cwdAbsPath, err := filepath.Abs(".")
	if err != nil {
		log.Fatalln(err)
	}

	handleCliFlags(systemUser.Name, cwdAbsPath)

	if helpFlag {
		fmt.Printf(help)
		os.Exit(0)
	}

	conf := &config{
		NodeID:    nodeIDFlag,
		WebPort:   webPortFlag,
		FilePort:  filePortFlag,
		SharePath: pathFlag,
		Debug:     debugFlag,
	}

	serve(conf)

}

func handleCliFlags(username, sharePath string) {
	flag.StringVar(&nodeIDFlag, "n", username, "node identifier")
	flag.StringVar(&pathFlag, "p", sharePath, "path to share files")
	flag.IntVar(&webPortFlag, "w", 1312, "port for web server")
	flag.IntVar(&filePortFlag, "s", 5555, "port for file server")
	flag.BoolVar(&debugFlag, "d", false, "debug logging")
	flag.BoolVar(&helpFlag, "h", false, "output help")
	flag.Parse()
}

type config struct {
	NodeID    string `json:"nodeID"`
	WebPort   int    `json:"webPort"`
	FilePort  int    `json:"filePort"`
	SharePath string `json:"sharePath"`
	Debug     bool   `json:"debug"`
}

type announcer struct {
	conf *config
}

type nodeInfo struct {
	NodeID        string `json:"nodeID"`
	Addr          string `json:"addr"`
	WebPort       int    `json:"webPort"`
	LastMulticast int64  `json:"lastMulticast"`
}

var (
	nodeMutex sync.Mutex
)

func newAnnouncePacket(n *nodeInfo) (string, error) {
	jsonMessage, err := json.Marshal(n)
	if err != nil {
		return "", err
	}

	message := fmt.Sprintf("%s%s%s", header, nodeAnnounceCommand, jsonMessage)

	return message, nil
}

func announceNode(nodeInfo *nodeInfo) {
	address, err := net.ResolveUDPAddr("udp", multicastAddress)
	if err != nil {
		return
	}

	conn, err := net.DialUDP("udp", nil, address)
	if err != nil {
		return
	}

	for {
		fmt.Println("sending multicast info")

		message, err := newAnnouncePacket(nodeInfo)
		if err != nil {
			fmt.Println("Could not get announce package")
			fmt.Println(err)
			continue
		}

		conn.Write([]byte(message))
		time.Sleep(announceIntervalSec * time.Second)
	}
}

const (
	multicastAddress    = "239.6.6.6:1337"
	multicastBufferSize = 4096
	nodeAnnounceCommand = "\x01"
	header              = "\x60\x0D\xF0\x0D"
	minPackageSize      = 6
	expireTimeoutSec    = 50
	announceIntervalSec = 10
)

func announcedNodeHandler(ninfo *nodeInfo, nodeList *list.List) {
	nodeMutex.Lock()
	updateNodeList(ninfo, nodeList)
	nodeMutex.Unlock()

	fmt.Println("Printing nodes")

	fmt.Print("[")
	for el := nodeList.Front(); el != nil; el = el.Next() {
		fmt.Print(el.Value.(*nodeInfo).NodeID, " ")
	}
	fmt.Print("]\n\n")
}

func updateNodeList(ninfo *nodeInfo, nodeList *list.List) {
	nodeExists := false
	for el := nodeList.Front(); el != nil; el = el.Next() {
		tmp := el.Value.(*nodeInfo)

		// Already in list
		if tmp.NodeID == ninfo.NodeID {
			tmp.LastMulticast = time.Now().Unix()
			fmt.Printf("Updating node %s multicast\n", ninfo.NodeID)
			nodeExists = true
			break
		}

	}

	for el := nodeList.Front(); el != nil; el = el.Next() {
		tmp := el.Value.(*nodeInfo)
		if isNodeExpired(tmp, expireTimeoutSec) {
			fmt.Println("Node expired, removing: ", tmp.NodeID)
			nodeList.Remove(el)
		}
	}

	if !nodeExists {
		fmt.Printf("Adding new node! %p %s\n", ninfo, ninfo.NodeID)
		ninfo.LastMulticast = time.Now().Unix()
		nodeList.PushBack(ninfo)
	}
}

func isNodeExpired(nodeInfo *nodeInfo, timeout int) bool {
	diff := time.Now().Unix() - nodeInfo.LastMulticast
	return diff > int64(timeout)
}

func parseAnnouncePacket(size int, addr *net.UDPAddr, packet []byte) (*nodeInfo, error) {
	if size <= minPackageSize {
		return nil, fmt.Errorf("Invalid packet size")
	}

	if strings.Compare(string(packet[0:len(header)]), header) != 0 {
		return nil, fmt.Errorf("Invalid packet header")
	}

	if string(packet[len(header):len(header)+1]) != nodeAnnounceCommand[0:] {
		return nil, fmt.Errorf("Command different than nodeAnnounceCommand")
	}

	fmt.Println("Packet command is nodeAnnounceCommand")

	payload := string(packet[len(header)+1:])
	payload = strings.Trim(payload, "\x00")

	nodeInfo := &nodeInfo{}

	err := json.Unmarshal([]byte(payload), nodeInfo)
	nodeInfo.Addr = addr.IP.String()
	nodeInfo.NodeID = fmt.Sprintf("%s", nodeInfo.NodeID)
	if err != nil {
		return nil, err
	}

	return nodeInfo, nil
}

func listenForNodes(nodeList *list.List) {
	address, err := net.ResolveUDPAddr("udp", multicastAddress)
	if err != nil {
		return
	}

	conn, err := net.ListenMulticastUDP("udp", nil, address)
	if err != nil {
		return
	}

	conn.SetReadBuffer(multicastBufferSize)

	for {
		packet := make([]byte, multicastBufferSize)
		size, udpAddr, err := conn.ReadFromUDP(packet)
		if err != nil {
			fmt.Println(err)
			continue
		}

		nodeInfo, err := parseAnnouncePacket(size, udpAddr, packet)

		if err != nil {
			fmt.Println(err)
			continue
		}
		fmt.Printf("Received multicast packet from %s Id: %s\n", udpAddr.String(), nodeInfo.NodeID)

		go announcedNodeHandler(nodeInfo, nodeList)
	}
}

func (a *announcer) Start(nodeList *list.List) {
	nodeInfo := &nodeInfo{
		NodeID:        a.conf.NodeID,
		Addr:          "",
		WebPort:       a.conf.WebPort,
		LastMulticast: 0,
	}

	go announceNode(nodeInfo)
	go listenForNodes(nodeList)
}

func startAnnouncer(conf *config, nodeList *list.List) {
	announcer := &announcer{conf: conf}
	announcer.Start(nodeList)
}

func serve(conf *config) {
	nodeList := list.New()

	go startAnnouncer(conf, nodeList)

	go fileServe(conf)
	go dashboardServe(conf, nodeList)

	for {
		// TODO: do this context cancel trick to escape cleanly?
		time.Sleep(time.Minute * 15)
	}
}

func fileServe(conf *config) {
	fileMux := http.NewServeMux()
	fileMux.Handle("/", http.FileServer(http.Dir(conf.SharePath)))
	http.ListenAndServe(fmt.Sprintf("0.0.0.0:%v", conf.WebPort), fileMux)
}

func configHandler(conf *config) func(w http.ResponseWriter, r *http.Request) {
	return func(w http.ResponseWriter, r *http.Request) {
		data, err := json.Marshal(conf)
		if err != nil {
			w.WriteHeader(http.StatusInternalServerError)
			return
		}

		w.Header().Add("Content-Type", "application/json")
		w.Write(data)
	}
}

func nodesHandler(nodeList *list.List) func(http.ResponseWriter, *http.Request) {
	return func(w http.ResponseWriter, r *http.Request) {
		nodes := make([]*nodeInfo, 0)
		for el := nodeList.Front(); el != nil; el = el.Next() {
			tmp := el.Value.(*nodeInfo)
			nodes = append(nodes, tmp)
		}

		data, err := json.Marshal(nodes)
		if err != nil {
			w.WriteHeader(http.StatusInternalServerError)
			return
		}

		w.Header().Add("Content-Type", "application/json")
		w.Write(data)
	}
}

func dashboardServe(conf *config, nodeList *list.List) {
	dashboardMux := http.NewServeMux()
	dashboardMux.Handle("/", http.FileServer(assets.AssetFS()))
	dashboardMux.HandleFunc("/api/config", configHandler(conf))
	dashboardMux.HandleFunc("/api/nodes", nodesHandler(nodeList))

	address := "0.0.0.0"

	fmt.Printf("Starting dashboard at %s:%v\n", address, conf.FilePort)
	err := http.ListenAndServe(fmt.Sprintf("%s:%v", address, conf.FilePort), dashboardMux)
	if err != nil {
		log.Fatal(err)
	}
}