// Test closing and opening of Host on same address
func TestHostClose(t *testing.T) {
	defer log.AfterTest(t)
	time.Sleep(time.Second)
	h1 := sda.NewLocalHost(2000)
	h2 := sda.NewLocalHost(2001)
	h1.ListenAndBind()
	_, err := h2.Connect(h1.ServerIdentity)
	if err != nil {
		t.Fatal("Couldn't Connect()", err)
	}
	err = h1.Close()
	if err != nil {
		t.Fatal("Couldn't close:", err)
	}
	err = h2.Close()
	if err != nil {
		t.Fatal("Couldn't close:", err)
	}
	log.Lvl3("Finished first connection, starting 2nd")
	h3 := sda.NewLocalHost(2002)
	h3.ListenAndBind()
	c, err := h2.Connect(h3.ServerIdentity)
	if err != nil {
		t.Fatal(h2, "Couldn Connect() to", h3)
	}
	log.Lvl3("Closing h3")
	err = h3.Close()
	if err != nil {
		// try closing the underlying connection manually and fail
		c.Close()
		t.Fatal("Couldn't Close()", h3)
	}
}

// handleAnnounce receive the announcement from another node
// it reply with an ACK.
func (b *Broadcast) handleContactNodes(msg struct {
	*sda.TreeNode
	ContactNodes
}) {
	log.Lvl3(b.Info(), "Received message from", msg.TreeNode.String())
	if msg.TreeNode.ID == b.Root().ID {
		b.repliesLeft = len(b.Tree().List()) - b.tnIndex - 1
		if b.repliesLeft == 0 {
			log.Lvl3("Won't contact anybody - finishing")
			b.SendTo(b.Root(), &Done{})
			return
		}
		log.Lvl3(b.Info(), "Contacting nodes:", b.repliesLeft)
		// Connect to all nodes that are later in the TreeNodeList, but only if
		// the message comes from root
		for _, tn := range b.Tree().List()[b.tnIndex+1:] {
			log.Lvl3("Connecting to", tn.String())
			err := b.SendTo(tn, &ContactNodes{})
			if err != nil {
				return
			}
		}
	} else {
		// Tell the caller we're done
		log.Lvl3("Sending back to", msg.TreeNode.ServerIdentity.String())
		b.SendTo(msg.TreeNode, &Done{})
	}
}

// Start will execute one cothority-binary for each server
// configured
func (d *Localhost) Start(args ...string) error {
	if err := os.Chdir(d.runDir); err != nil {
		return err
	}
	log.Lvl4("Localhost: chdir into", d.runDir)
	ex := d.runDir + "/" + d.Simulation
	d.running = true
	log.Lvl1("Starting", d.servers, "applications of", ex)
	for index := 0; index < d.servers; index++ {
		d.wgRun.Add(1)
		log.Lvl3("Starting", index)
		host := "localhost" + strconv.Itoa(index)
		cmdArgs := []string{"-address", host, "-monitor",
			"localhost:" + strconv.Itoa(d.monitorPort),
			"-simul", d.Simulation,
			"-debug", strconv.Itoa(log.DebugVisible()),
		}
		cmdArgs = append(args, cmdArgs...)
		log.Lvl3("CmdArgs are", cmdArgs)
		cmd := exec.Command(ex, cmdArgs...)
		cmd.Stdout = os.Stdout
		cmd.Stderr = os.Stderr
		go func(i int, h string) {
			log.Lvl3("Localhost: will start host", h)
			err := cmd.Run()
			if err != nil {
				log.Error("Error running localhost", h, ":", err)
				d.errChan <- err
			}
			d.wgRun.Done()
			log.Lvl3("host (index", i, ")", h, "done")
		}(index, host)
	}
	return nil
}

// Wait for all processes to finish
func (d *Localhost) Wait() error {
	log.Lvl3("Waiting for processes to finish")

	var err error
	go func() {
		d.wgRun.Wait()
		log.Lvl3("WaitGroup is 0")
		// write to error channel when done:
		d.errChan <- nil
	}()

	// if one of the hosts fails, stop waiting and return the error:
	select {
	case e := <-d.errChan:
		log.Lvl3("Finished waiting for hosts:", e)
		if e != nil {
			if err := d.Cleanup(); err != nil {
				log.Error("Couldn't cleanup running instances",
					err)
			}
			err = e
		}
	}

	log.Lvl2("Processes finished")
	return err
}

// Filter out a serie of values
func (df *DataFilter) Filter(measure string, values []float64) []float64 {
	// do we have a filter for this measure ?
	if _, ok := df.percentiles[measure]; !ok {
		return values
	}
	// Compute the percentile value
	max, err := stats.PercentileNearestRank(values, df.percentiles[measure])
	if err != nil {
		log.Lvl2("Monitor: Error filtering data(", values, "):", err)
		return values
	}

	// Find the index from where to filter
	maxIndex := -1
	for i, v := range values {
		if v > max {
			maxIndex = i
		}
	}
	// check if we foud something to filter out
	if maxIndex == -1 {
		log.Lvl3("Filtering: nothing to filter for", measure)
		return values
	}
	// return the values below the percentile
	log.Lvl3("Filtering: filters out", measure, ":", maxIndex, "/", len(values))
	return values[:maxIndex]
}

// handleCommit receives commit messages and signal the end if it received
// enough of it.
func (p *Protocol) handleCommit(com *Commit) {
	if p.state != stateCommit {
		//	log.Lvl3(p.Name(), "STORE handle commit packet")
		p.tempCommitMsg = append(p.tempCommitMsg, com)
		return
	}
	// finish after threshold of Commit msgs
	p.commitMsgCount++
	log.Lvl4(p.Name(), "----------------\nWe got", p.commitMsgCount,
		"COMMIT msgs and threshold is", p.threshold)
	if p.IsRoot() {
		log.Lvl4("Leader got ", p.commitMsgCount)
	}
	if p.commitMsgCount >= p.threshold {
		p.state = stateFinished
		// reset counter
		p.commitMsgCount = 0
		log.Lvl3(p.Name(), "Threshold reached: We are done... CONSENSUS")
		if p.IsRoot() && p.onDoneCB != nil {
			log.Lvl3(p.Name(), "We are root and threshold reached: return to the simulation.")
			p.onDoneCB()
			p.finish()
		}
		return
	}
}

// listen starts listening for messages coming from any host that tries to
// contact this host. If 'wait' is true, it will try to connect to itself before
// returning.
func (h *Host) listen(wait bool) {
	log.Lvl3(h.ServerIdentity.First(), "starts to listen")
	fn := func(c network.SecureConn) {
		log.Lvl3(h.workingAddress, "Accepted Connection from", c.Remote())
		// register the connection once we know it's ok
		h.registerConnection(c)
		h.handleConn(c)
	}
	go func() {
		log.Lvl4("Host listens on:", h.workingAddress)
		err := h.host.Listen(fn)
		if err != nil {
			log.Fatal("Couldn't listen on", h.workingAddress, ":", err)
		}
	}()
	if wait {
		for {
			log.Lvl4(h.ServerIdentity.First(), "checking if listener is up")
			_, err := h.Connect(h.ServerIdentity)
			if err == nil {
				log.Lvl4(h.ServerIdentity.First(), "managed to connect to itself")
				break
			}
			time.Sleep(network.WaitRetry)
		}
	}
}

// CreateRoster creates an Roster with the host-names in 'addresses'.
// It creates 's.Hosts' entries, starting from 'port' for each round through
// 'addresses'
func (s *SimulationBFTree) CreateRoster(sc *SimulationConfig, addresses []string, port int) {
	start := time.Now()
	nbrAddr := len(addresses)
	if sc.PrivateKeys == nil {
		sc.PrivateKeys = make(map[string]abstract.Scalar)
	}
	hosts := s.Hosts
	if s.SingleHost {
		// If we want to work with a single host, we only make one
		// host per server
		log.Fatal("Not supported yet")
		hosts = nbrAddr
		if hosts > s.Hosts {
			hosts = s.Hosts
		}
	}
	localhosts := false
	listeners := make([]net.Listener, hosts)
	if strings.Contains(addresses[0], "localhost") {
		localhosts = true
	}
	entities := make([]*network.ServerIdentity, hosts)
	log.Lvl3("Doing", hosts, "hosts")
	key := config.NewKeyPair(network.Suite)
	for c := 0; c < hosts; c++ {
		key.Secret.Add(key.Secret,
			key.Suite.Scalar().One())
		key.Public.Add(key.Public,
			key.Suite.Point().Base())
		address := addresses[c%nbrAddr] + ":"
		if localhosts {
			// If we have localhosts, we have to search for an empty port
			var err error
			listeners[c], err = net.Listen("tcp", ":0")
			if err != nil {
				log.Fatal("Couldn't search for empty port:", err)
			}
			_, p, _ := net.SplitHostPort(listeners[c].Addr().String())
			address += p
			log.Lvl4("Found free port", address)
		} else {
			address += strconv.Itoa(port + c/nbrAddr)
		}
		entities[c] = network.NewServerIdentity(key.Public, address)
		sc.PrivateKeys[entities[c].Addresses[0]] = key.Secret
	}
	// And close all our listeners
	if localhosts {
		for _, l := range listeners {
			err := l.Close()
			if err != nil {
				log.Fatal("Couldn't close port:", l, err)
			}
		}
	}

	sc.Roster = NewRoster(entities)
	log.Lvl3("Creating entity List took: " + time.Now().Sub(start).String())
}

// EnableMeasure will actually allow the sending of the measures if given true.
// Otherwise all measures won't be sent at all.
func EnableMeasure(b bool) {
	if b {
		log.Lvl3("Monitor: Measure enabled")
	} else {
		log.Lvl3("Monitor: Measure disabled")
	}
	enabled = b
}

func TestHostClose2(t *testing.T) {
	defer log.AfterTest(t)
	local := sda.NewLocalTest()
	defer local.CloseAll()

	_, _, tree := local.GenTree(2, false, true, true)
	log.Lvl3(tree.Dump())
	time.Sleep(time.Millisecond * 100)
	log.Lvl3("Done")
}

// CreateTree the tree as defined in SimulationBFTree and stores the result
// in 'sc'
func (s *SimulationBFTree) CreateTree(sc *SimulationConfig) error {
	log.Lvl3("CreateTree strarted")
	start := time.Now()
	if sc.Roster == nil {
		return errors.New("Empty Roster")
	}
	sc.Tree = sc.Roster.GenerateBigNaryTree(s.BF, s.Hosts)
	log.Lvl3("Creating tree took: " + time.Now().Sub(start).String())
	return nil
}

// RunTest a single test - takes a test-file as a string that will be copied
// to the deterlab-server
func RunTest(rc platform.RunConfig) (*monitor.Stats, error) {
	done := make(chan struct{})
	CheckHosts(rc)
	rc.Delete("simulation")
	rs := monitor.NewStats(rc.Map(), "hosts", "bf")
	monitor := monitor.NewMonitor(rs)

	if err := deployP.Deploy(rc); err != nil {
		log.Error(err)
		return rs, err
	}

	monitor.SinkPort = monitorPort
	if err := deployP.Cleanup(); err != nil {
		log.Error(err)
		return rs, err
	}
	monitor.SinkPort = monitorPort
	go func() {
		if err := monitor.Listen(); err != nil {
			log.Fatal("Could not monitor.Listen():", err)
		}
	}()
	// Start monitor before so ssh tunnel can connect to the monitor
	// in case of deterlab.
	err := deployP.Start()
	if err != nil {
		log.Error(err)
		return rs, err
	}

	go func() {
		var err error
		if err = deployP.Wait(); err != nil {
			log.Lvl3("Test failed:", err)
			if err := deployP.Cleanup(); err != nil {
				log.Lvl3("Couldn't cleanup platform:", err)
			}
			done <- struct{}{}
		}
		log.Lvl3("Test complete:", rs)
		done <- struct{}{}
	}()

	timeOut := getRunWait(rc)
	// can timeout the command if it takes too long
	select {
	case <-done:
		monitor.Stop()
		return rs, nil
	case <-time.After(time.Second * time.Duration(timeOut)):
		monitor.Stop()
		return rs, errors.New("Simulation timeout")
	}
}

// FuncPrepareClose sends a `PrepareClose`-message down the tree.
func (p *ProtocolCloseAll) FuncPrepareClose(pc PrepareCloseMsg) {
	log.Lvl3(pc.ServerIdentity.Addresses, "sent PrepClose to", p.ServerIdentity().Addresses)
	if !p.IsLeaf() {
		for _, c := range p.Children() {
			err := p.SendTo(c, &PrepareClose{})
			log.Lvl3(p.ServerIdentity().Addresses, "sends to", c.ServerIdentity.Addresses, "(err=", err, ")")
		}
	} else {
		p.FuncClose(nil)
	}
}

// Cleanup kills all running cothority-binaryes
func (d *Localhost) Cleanup() error {
	log.Lvl3("Cleaning up")
	ex := d.runDir + "/" + d.Simulation
	err := exec.Command("pkill", "-f", ex).Run()
	if err != nil {
		log.Lvl3("Error stopping localhost", err)
	}

	// Wait for eventual connections to clean up
	time.Sleep(time.Second)
	return nil
}

// Configure various internal variables
func (d *Localhost) Configure(pc *Config) {
	pwd, _ := os.Getwd()
	d.runDir = pwd + "/platform/localhost"
	d.localDir = pwd
	d.debug = pc.Debug
	d.running = false
	d.monitorPort = pc.MonitorPort
	d.errChan = make(chan error)
	if d.Simulation == "" {
		log.Fatal("No simulation defined in simulation")
	}
	log.Lvl3(fmt.Sprintf("Localhost dirs: RunDir %s", d.runDir))
	log.Lvl3("Localhost configured ...")
}

// HandleReply is the message going up the tree and holding a counter
// to verify the number of nodes.
func (p *ProtocolExampleHandlers) HandleReply(reply []StructReply) error {
	children := 1
	for _, c := range reply {
		children += c.ChildrenCount
	}
	log.Lvl3(p.ServerIdentity().Addresses, "is done with total of", children)
	if !p.IsRoot() {
		log.Lvl3("Sending to parent")
		return p.SendTo(p.Parent(), &Reply{children})
	}
	log.Lvl3("Root-node is done - nbr of children found:", children)
	p.ChildCount <- children
	return nil
}

// listen will select on the differents channels
func (nt *Ntree) listen() {
	for {
		select {
		// Dispatch the block through the whole tree
		case msg := <-nt.announceChan:
			log.Lvl3(nt.Name(), "Received Block announcement")
			nt.block = msg.BlockAnnounce.Block
			// verify the block
			go byzcoin.VerifyBlock(nt.block, "", "", nt.verifyBlockChan)
			if nt.IsLeaf() {
				nt.startBlockSignature()
				continue
			}
			for _, tn := range nt.Children() {
				err := nt.SendTo(tn, &msg.BlockAnnounce)
				if err != nil {
					log.Error(nt.Name(),
						"couldn't send to", tn.Name(),
						err)
				}
			}
			// generate your own signature / exception and pass that up to the
			// root
		case msg := <-nt.blockSignatureChan:
			nt.handleBlockSignature(&msg.NaiveBlockSignature)
			// Dispatch the signature + expcetion made before through the whole
			// tree
		case msg := <-nt.roundSignatureRequestChan:
			log.Lvl3(nt.Name(), " Signature Request Received")
			go nt.verifySignatureRequest(&msg.RoundSignatureRequest)

			if nt.IsLeaf() {
				nt.startSignatureResponse()
				continue
			}

			for _, tn := range nt.Children() {
				err := nt.SendTo(tn, &msg.RoundSignatureRequest)
				if err != nil {
					log.Error(nt.Name(), "couldn't sent to",
						tn.Name(), err)
				}
			}
			// Decide if we want to sign this or not
		case msg := <-nt.roundSignatureResponseChan:
			nt.handleRoundSignatureResponse(&msg.RoundSignatureResponse)
		}
	}
}

// ConnectSink connects to the given endpoint and initialises a json
// encoder. It can be the address of a proxy or a monitoring process.
// Returns an error if it could not connect to the endpoint.
func ConnectSink(addr string) error {
	if encoder != nil {
		return nil
	}
	log.Lvl3("Connecting to:", addr)
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		return err
	}
	log.Lvl3("Connected to sink:", addr)
	sink = addr
	connection = conn
	encoder = json.NewEncoder(conn)
	return nil
}

// Dispatch can handle timeouts
func (p *Propagate) Dispatch() error {
	process := true
	log.Lvl4(p.ServerIdentity())
	for process {
		p.Lock()
		timeout := time.Millisecond * time.Duration(p.sd.Msec)
		p.Unlock()
		select {
		case msg := <-p.ChannelSD:
			log.Lvl3(p.ServerIdentity(), "Got data from", msg.ServerIdentity)
			if p.onData != nil {
				_, netMsg, err := network.UnmarshalRegistered(msg.Data)
				if err == nil {
					p.onData(netMsg)
				}
			}
			if !p.IsRoot() {
				log.Lvl3(p.ServerIdentity(), "Sending to parent")
				p.SendToParent(&PropagateReply{})
			}
			if p.IsLeaf() {
				process = false
			} else {
				log.Lvl3(p.ServerIdentity(), "Sending to children")
				p.SendToChildren(&msg.PropagateSendData)
			}
		case <-p.ChannelReply:
			p.received++
			log.Lvl4(p.ServerIdentity(), "received:", p.received, p.subtree)
			if !p.IsRoot() {
				p.SendToParent(&PropagateReply{})
			}
			if p.received == p.subtree {
				process = false
			}
		case <-time.After(timeout):
			log.Fatal("Timeout")
			process = false
		}
	}
	if p.IsRoot() {
		if p.onDoneCb != nil {
			p.onDoneCb(p.received + 1)
		}
	}
	p.Done()
	return nil
}

// NewDataFilter returns a new data filter initialized with the rights values
// taken out from the run config. If absent, will take defaults values.
// Keys expected are:
// discard_measurementname = perc => will take the lower and upper percentile =
// perc
// discard_measurementname = lower,upper => will take different percentiles
func NewDataFilter(config map[string]string) DataFilter {
	df := DataFilter{
		percentiles: make(map[string]float64),
	}
	reg, err := regexp.Compile("filter_(\\w+)")
	if err != nil {
		log.Lvl1("DataFilter: Error compiling regexp:", err)
		return df
	}
	// analyse the each entry
	for k, v := range config {
		if measure := reg.FindString(k); measure == "" {
			continue
		} else {
			// this value must be filtered by how many ?
			perc, err := strconv.ParseFloat(v, 64)
			if err != nil {
				log.Lvl1("DataFilter: Cannot parse value for filter measure:", measure)
				continue
			}
			measure = strings.Replace(measure, "filter_", "", -1)
			df.percentiles[measure] = perc
		}
	}
	log.Lvl3("Filtering:", df.percentiles)
	return df
}