var globalNetperfRX netperfStats

// Used for collecting stats for netperf
type netperfStats struct {
	received  uint64
	id        uint64
	first     int64
	last      int64
	connected int32
	mu        sync.Mutex
}

// recorder will return a counter linked to the id.
// If the ID is unknown and no other counters are running
// a new will be initialized.
// The returned pointer must be atomically incremented.
// If nil is returned another recording is already.
// Whenever a non-nil value is returned release() must be called.
func (n *netperfStats) recorder(id uint64) *uint64 {
	n.mu.Lock()
	defer n.mu.Unlock()
	if id == n.id {
		atomic.AddInt32(&n.connected, 1)
		return &n.received
	}
	if atomic.LoadInt32(&n.connected) > 0 {
		return nil
	}
	// Replace with new, no-one is connected.
	n.id = id
	n.received = 0
	n.connected++
	n.first = time.Now().UnixNano()
	n.last = 0
	return &n.received
}

// getRecorded returns statistics recorded on the specific ID.
func (n *netperfStats) getRecorded(id uint64) (uint64, time.Duration, error) {
	n.mu.Lock()
	defer n.mu.Unlock()
	if n.id != id {
		return 0, 0, errors.New("id not found")
	}
	if n.connected > 0 {
		return 0, 0, errors.New("clients are still connected")
	}
	return n.received, time.Duration(n.last-n.first) * time.Nanosecond, nil
}

// release a connection from the stats.
func (n *netperfStats) release() {
	atomic.AddInt32(&n.connected, -1)
	atomic.StoreInt64(&n.last, time.Now().UnixNano())
}

// DevNull - everything goes to io.Discard
func (s *peerRESTServer) DevNull(w http.ResponseWriter, r *http.Request) {
	if !s.IsValid(w, r) {
		s.writeErrorResponse(w, errors.New("invalid request"))
		return
	}
	id, err := strconv.ParseUint(r.Form.Get("id"), 10, 64)
	if err != nil {
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	rx := globalNetperfRX.recorder(id)
	if rx == nil {
		w.WriteHeader(http.StatusConflict)
		return
	}
	defer globalNetperfRX.release()
	for {
		n, err := io.CopyN(io.Discard, r.Body, 128*humanize.KiByte)
		atomic.AddUint64(rx, uint64(n))
		if err != nil {
			break
		}
	}
}

var globalNetperfRX netperfStats

// Used for collecting stats for netperf
type netperfStats struct {
	id               uint64 // id of current test
	rxCounter        uint64 // temporary counter.
	received         uint64 // received between lastConnected and firstDisconnect
	lastConnected    int64  // time when last client connected.
	firstDisconnect  int64  // time when first client disconnected.
	connected        int32  // n connected.
	someDisconnected int32  // 0 when none have disconnected yet, 1 when any has disconnected.
	mu               sync.Mutex
}

// recorder will return a counter linked to the id.
// If the ID is unknown and no other counters are running
// a new will be initialized.
// The returned pointer must be atomically incremented.
// If nil is returned another recording is already.
// Whenever a non-nil value is returned release() must be called.
func (n *netperfStats) recorder(id uint64) *uint64 {
	n.mu.Lock()
	defer n.mu.Unlock()
	if id == n.id {
		atomic.AddInt32(&n.connected, 1)
		// Reset timing
		atomic.StoreUint64(&n.rxCounter, 0)
		atomic.StoreInt64(&n.lastConnected, time.Now().UnixNano())

		return &n.rxCounter
	}
	if atomic.LoadInt32(&n.connected) > 0 {
		return nil
	}
	// Replace with new, no-one is connected.
	n.id = id
	n.rxCounter = 0
	n.connected++
	n.lastConnected = time.Now().UnixNano()
	n.firstDisconnect = 0
	n.someDisconnected = 0
	return &n.rxCounter
}

// getRecorded returns statistics recorded on the specific ID.
func (n *netperfStats) getRecorded(id uint64) (uint64, time.Duration, error) {
	n.mu.Lock()
	defer n.mu.Unlock()
	if n.id != id {
		return 0, 0, errors.New("id not found")
	}
	if n.connected > 0 {
		return 0, 0, errors.New("clients are still connected")
	}
	return n.rxCounter, time.Duration(n.firstDisconnect-n.lastConnected) * time.Nanosecond, nil
}

// release a connection from the stats.
func (n *netperfStats) release() {
	atomic.AddInt32(&n.connected, -1)
	// Store when first disconnects.
	if atomic.CompareAndSwapInt32(&n.someDisconnected, 0, 1) {
		r := atomic.LoadUint64(&n.rxCounter)
		atomic.StoreUint64(&n.received, r)
		atomic.StoreInt64(&n.firstDisconnect, time.Now().UnixNano())
	}
}

// DevNull - everything goes to io.Discard
func (s *peerRESTServer) DevNull(w http.ResponseWriter, r *http.Request) {
	if !s.IsValid(w, r) {
		s.writeErrorResponse(w, errors.New("invalid request"))
		return
	}
	id, err := strconv.ParseUint(r.Form.Get("id"), 10, 64)
	if err != nil {
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	rx := globalNetperfRX.recorder(id)
	if rx == nil {
		w.WriteHeader(http.StatusConflict)
		return
	}
	defer globalNetperfRX.release()
	for {
		n, err := io.CopyN(io.Discard, r.Body, 128*humanize.KiByte)
		atomic.AddUint64(rx, uint64(n))
		if err != nil {
			break
		}
	}
}