// Return all entries in the chosen part.
func (ht *HashTable) GetPartition(partNum, partSize int) (keys, vals []int) {
	rangeStart, rangeEnd := GetPartitionRange(partNum, partSize)
	prealloc := (rangeEnd - rangeStart) * PER_BUCKET
	keys = make([]int, 0, prealloc)
	vals = make([]int, 0, prealloc)
	for head := rangeStart; head < rangeEnd; head++ {
		var entry, bucket int = 0, head
		for {
			entryAddr := bucket*BUCKET_SIZE + BUCKET_HEADER + entry*ENTRY_SIZE
			entryKey, _ := binary.Varint(ht.Buf[entryAddr+1 : entryAddr+11])
			entryVal, _ := binary.Varint(ht.Buf[entryAddr+11 : entryAddr+21])
			if ht.Buf[entryAddr] == 1 {
				keys = append(keys, int(entryKey))
				vals = append(vals, int(entryVal))
			} else if entryKey == 0 && entryVal == 0 {
				break
			}
			if entry++; entry == PER_BUCKET {
				entry = 0
				if bucket = ht.nextBucket(bucket); bucket == 0 {
					return
				}
			}
		}
	}
	return
}

func ParseMessage(b []byte) (m Message, err error) {
	err = nil
	i, _ := binary.Varint(b[:1])
	m.majorVer = int8(i) //	version-number            	2 bytes  	- required
	i, _ = binary.Varint(b[1:2])
	m.minorVer = int8(i)
	//log.Println(m)

	//panic(1)

	ii, _ := binary.Uvarint(b[3:4])
	m.operationIdStatusCode = uint16(ii) //	operation-id (request)		2 bytes  	- required
	m.IsResponse = typeCheck(uint16(ii)) //	or status-code (response)

	iii, _ := binary.Varint(b[4:8]) //	request-id 					4 bytes  	- required
	m.requestId = int32(iii)
	//et := bytes.IndexByte(b[8:], TAG_END)
	//log.Println("message:", m)
	ags := splitAValues(b[8:]) //et+9]) //	attribute-group				n bytes 	- 0 or more

	m.attributeGroups = ags

	// m.endAttributeTag = b[et] //	end-of-attributes-tag		1 byte   	- required

	//m.Data = b[et+8:] //	data q bytes  	- optional
	//log.Println("Data: ", m.Data)

	return
}

// Look up values by key.
func (ht *HashTable) Get(key, limit int) (vals []int) {
	if limit == 0 {
		vals = make([]int, 0, 10)
	} else {
		vals = make([]int, 0, limit)
	}
	for count, entry, bucket := 0, 0, HashKey(key); ; {
		entryAddr := bucket*BUCKET_SIZE + BUCKET_HEADER + entry*ENTRY_SIZE
		entryKey, _ := binary.Varint(ht.Buf[entryAddr+1 : entryAddr+11])
		entryVal, _ := binary.Varint(ht.Buf[entryAddr+11 : entryAddr+21])
		if ht.Buf[entryAddr] == 1 {
			if int(entryKey) == key {
				vals = append(vals, int(entryVal))
				if count++; count == limit {
					return
				}
			}
		} else if entryKey == 0 && entryVal == 0 {
			return
		}
		if entry++; entry == PER_BUCKET {
			entry = 0
			if bucket = ht.nextBucket(bucket); bucket == 0 {
				return
			}
		}
	}
}

//  Int64 decodes a int64 from buffer
func (d *Dec) Int64() int64 {
	if d.err != nil {
		return 0
	}
	if d.i >= len(d.decbuf) || d.i < 0 /*overflow*/ {
		d.err = errNoDecData
		return 0
	}
	d.lng = int(d.decbuf[d.i])
	// if d.lng <= 0 {
	// 	d.err = errDecode
	// 	return 0
	// }
	d.i++
	d.lst = d.i + d.lng
	if d.lst > len(d.decbuf) {
		d.err = errDecodeNotEnoughtData
		return 0
	}
	var x int64
	if d.lst == len(d.decbuf) {
		x, d.i = binary.Varint(d.decbuf[d.i:])
	} else {
		x, d.i = binary.Varint(d.decbuf[d.i:d.lst])
	}
	if d.i <= 0 {
		d.err = errDecode
		return 0
	}
	d.i = d.lst
	return x
}

func (b *binaryNomsReader) readNumber() Number {
	// b.assertCanRead(binary.MaxVarintLen64 * 2)
	i, count := binary.Varint(b.buff[b.offset:])
	b.offset += uint32(count)
	exp, count2 := binary.Varint(b.buff[b.offset:])
	b.offset += uint32(count2)
	return Number(intExpToFloat64(i, int(exp)))
}

func (t *nameWithLanguage) UnMarshalIPP(b []byte) error {
	l, _ := binary.Varint(b[:2])
	ll := int16(l)
	t.nameLength = signedShort(ll) // a. number of octets in the following field
	t.name = b[3:ll]               // b. type []byte
	lb, _ := binary.Varint(b[ll+1 : ll+3])
	t.valueLength = signedShort(lb) // c. the number of octets in the following field
	t.value = b[ll+4:]              // d. type []byte

	return nil
}

func UnmarshalIdRefsBunch2(buf []byte, idRefs []element.IdRefs) []element.IdRefs {
	length, n := binary.Uvarint(buf)
	if n <= 0 {
		return nil
	}

	offset := n

	if uint64(cap(idRefs)) < length {
		idRefs = make([]element.IdRefs, length)
	} else {
		idRefs = idRefs[:length]
	}

	last := int64(0)
	for i := 0; uint64(i) < length; i++ {
		idRefs[i].Id, n = binary.Varint(buf[offset:])
		if n <= 0 {
			panic("no data")
		}
		offset += n
		idRefs[i].Id += last
		last = idRefs[i].Id
	}
	var numRefs uint64
	for i := 0; uint64(i) < length; i++ {
		numRefs, n = binary.Uvarint(buf[offset:])
		if n <= 0 {
			panic("no data")
		}
		offset += n
		if uint64(cap(idRefs[i].Refs)) < numRefs {
			idRefs[i].Refs = make([]int64, numRefs)
		} else {
			idRefs[i].Refs = idRefs[i].Refs[:numRefs]
		}
	}
	last = 0
	for idIdx := 0; uint64(idIdx) < length; idIdx++ {
		for refIdx := 0; refIdx < len(idRefs[idIdx].Refs); refIdx++ {
			idRefs[idIdx].Refs[refIdx], n = binary.Varint(buf[offset:])
			if n <= 0 {
				panic("no data")
			}
			offset += n
			idRefs[idIdx].Refs[refIdx] += last
			last = idRefs[idIdx].Refs[refIdx]
		}
	}
	return idRefs
}

func UnmarshalDeltaNodes(buf []byte, nodes []element.Node) ([]element.Node, error) {
	length, n := binary.Uvarint(buf)
	if n <= 0 {
		return nil, varintErr
	}
	var offset = n

	if uint64(cap(nodes)) < length {
		nodes = make([]element.Node, length)
	} else {
		nodes = nodes[:length]
	}

	lastId := int64(0)
	for i := 0; uint64(i) < length; i++ {
		id, n := binary.Varint(buf[offset:])
		if n <= 0 {
			return nil, varintErr
		}
		offset += n
		id = lastId + id
		nodes[i].Id = id
		lastId = id
	}

	lastLong := int64(0)
	for i := 0; uint64(i) < length; i++ {
		long, n := binary.Varint(buf[offset:])
		if n <= 0 {
			return nil, varintErr
		}
		offset += n
		long = lastLong + long
		nodes[i].Long = IntToCoord(uint32(long))
		lastLong = long
	}

	lastLat := int64(0)
	for i := 0; uint64(i) < length; i++ {
		lat, n := binary.Varint(buf[offset:])
		if n <= 0 {
			return nil, varintErr
		}
		offset += n
		lat = lastLat + lat
		nodes[i].Lat = IntToCoord(uint32(lat))
		lastLat = lat
	}

	return nodes, nil
}

func peekVarint(b []byte) (int, error) {
	_, n := binary.Varint(b)
	if n < 0 {
		return 0, errors.New("value larger than 64 bits")
	}
	return n, nil
}

// Record each message's latency. The message contains the timestamp when it was sent.
// If it's the last message, compute the average latency and print it out. Return true
// if the message is the last one, otherwise return false.
func (handler *LatencyMessageHandler) ReceiveMessage(message []byte) bool {
	now := time.Now().UnixNano()
	then, _ := binary.Varint(message)

	// TODO: Figure out why nanomsg and ZeroMQ sometimes receive empty messages.
	if then != 0 {
		handler.Latencies = append(handler.Latencies, (float32(now-then))/1000/1000)
	}

	handler.messageCounter++
	if handler.messageCounter == handler.NumberOfMessages {
		sum := float32(0)
		for _, latency := range handler.Latencies {
			sum += latency
		}
		avgLatency := float32(sum) / float32(len(handler.Latencies))
		log.Printf("Mean latency for %d messages: %f ms\n", handler.NumberOfMessages,
			avgLatency)

		handler.completionLock.Lock()
		handler.hasCompleted = true
		handler.completionLock.Unlock()

		return true
	}

	return false
}

// Overwrite or re-insert a document, return the new document ID if re-inserted.
func (col *Collection) Update(id int, data []byte) (newID int, err error) {
	dataLen := len(data)
	if dataLen > DOC_MAX_ROOM {
		return 0, dberr.New(dberr.ErrorDocTooLarge, DOC_MAX_ROOM, dataLen)
	}
	if id < 0 || id >= col.Used-DOC_HEADER || col.Buf[id] != 1 {
		return 0, dberr.New(dberr.ErrorNoDoc, id)
	}
	currentDocRoom, _ := binary.Varint(col.Buf[id+1 : id+11])
	if currentDocRoom > DOC_MAX_ROOM {
		return 0, dberr.New(dberr.ErrorNoDoc, id)
	}
	if docEnd := id + DOC_HEADER + int(currentDocRoom); docEnd >= col.Size {
		return 0, dberr.New(dberr.ErrorNoDoc, id)
	}
	if dataLen <= int(currentDocRoom) {
		padding := id + DOC_HEADER + len(data)
		paddingEnd := id + DOC_HEADER + int(currentDocRoom)
		// Overwrite data and then overwrite padding
		copy(col.Buf[id+DOC_HEADER:padding], data)
		for ; padding < paddingEnd; padding += LEN_PADDING {
			copySize := LEN_PADDING
			if padding+LEN_PADDING >= paddingEnd {
				copySize = paddingEnd - padding
			}
			copy(col.Buf[padding:padding+copySize], PADDING)
		}
		return id, nil
	}

	// No enough room - re-insert the document
	col.Delete(id)
	return col.Insert(data)
}

// authorizeConnection establishes the RCON connection to the server addr
// using password for authorization.
func authorizeConnection(addr, password string) (net.Conn, error) {
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		return nil, err
	}

	id := 0
	conn.Write(rconPacket(id, 3, password))

	var buf [4096]byte

	// serverdata_response_value
	conn.SetReadDeadline(time.Now().Add(5000 * time.Millisecond))
	_, err = conn.Read(buf[0:])
	if err != nil {
		return nil, errors.New("Authorization failed: " + err.Error())
	}

	// serverdata_auth_response
	_, err = conn.Read(buf[0:])
	conn.SetReadDeadline(time.Now().Add(15000 * time.Millisecond))
	if err != nil {
		return nil, errors.New("Authorization failed: " + err.Error())
	}

	resID, _ := binary.Varint(buf[4:7])
	if int(resID) != id {
		return nil, errors.New("Authorization failed: invalid RCON password")
	}

	return conn, nil
}

func (ck *DBClerk) QuietGetInt(table Table, key string) (int, bool) {
	table_key := append([]byte{byte(table)}, []byte(key)...)
	bytes, err := ck.db.Get(table_key, nil)

	val, _ := binary.Varint(bytes)
	return int(val), (err == nil)
}

func (c *Integers) UnmarshalBinary(b []byte) error {
	if len(b) < 1 {
		return fmt.Errorf("compare: missing type byte")
	}
	if b[0] != byte(TypeIntegers) {
		return fmt.Errorf("compare: expected type %d, got %d", TypeIntegers, b[0])
	}
	b = b[1:]
	*c = Integers{}
	for len(b) > 0 {
		if len(b) < 2 {
			return fmt.Errorf("compare: unexpected end of buffer decoding integer")
		}
		op := IntegerOp(b[0])
		if op > integerOpMax {
			return fmt.Errorf("compare: unknown integer operation %d", b[0])
		}
		i, n := binary.Varint(b[1:])
		if n <= 0 {
			return fmt.Errorf("compare: invalid integer")
		}
		*c = append(*c, Integer{Op: op, Int: i})
		b = b[n+1:]
	}
	return nil
}

func ListenToBroadcast(listenAddress string, receiveCh, timeoutCh chan int) {

	ServerAddress, err := net.ResolveUDPAddr("udp", listenAddress)
	CheckError(err)

	ServerConnection, err := net.ListenUDP("udp", ServerAddress)
	CheckError(err)

	defer ServerConnection.Close()

	var count int64

mainloop:
	for {
		buffer := make([]byte, 64)
		ServerConnection.SetDeadline(time.Now().Add(3000 * time.Millisecond))
		n, _, err := ServerConnection.ReadFromUDP(buffer)
		CheckError(err)
		count, _ = binary.Varint(buffer[0:n])
		c := int(count)
		receiveCh <- c

		if err != nil {
			timeoutCh <- 1
			break mainloop
		}

	}
}

func TestBytesToInt(t *testing.T) {
	test1 := int64(256)
	buf := new(bytes.Buffer)
	binary.Write(buf, binary.LittleEndian, test1)
	//binary.PutVarint(buf.Bytes(), int64(test1))
	fmt.Println("byte is ", buf.Bytes())
	bufr := bytes.NewBuffer(buf.Bytes())
	var rest int64
	err := binary.Read(bufr, binary.LittleEndian, &rest)
	if err != nil {
		t.Error(err)
	}
	fmt.Println("rest value ", rest)

	num, _ := binary.Varint(buf.Bytes())
	fmt.Println("num is ", num)

	var testS []TableId
	testS = append(testS, 34, 3443, 344242)
	tids := GetSliceBytesFromInts(testS)
	fmt.Println("length is ", len(tids), "val is ", tids)

	tidr := GetSliceIntFromBytes(tids)
	fmt.Println("int value is ", tidr)

	t4 := ToInt64FromBytes(tids[0:8])
	fmt.Println("t4 val is ", t4)

}

func getGoVarintLen(b []byte) (int, int, error) {
	i, n := binary.Varint(b)
	if n <= 0 {
		return 0, 0, util.Errorf("invalid varint")
	}
	return n, int(i), nil
}

func (b Bucket) Get(key string) int64 {
	buff := b.Bucket.Get([]byte(key))
	if val, n := binary.Varint(buff); n > 0 {
		return val
	}
	return 0
}

// DecodeToken will take a token as sent by a client and translate it into a
// key to look up the full token on the server side and the raw secret.
func decodeToken(ctx context.Context, token string) (*datastore.Key, []byte, error) {
	b, err := hex.DecodeString(token)
	if err != nil {
		return nil, nil, err
	}

	if len(b) != len(kinds)*8+tokenLength {
		return nil, nil, errors.New("token length mismatch")
	}

	var intIDs [len(kinds)]int64
	for i := range intIDs {
		intID, n := binary.Varint(b[i*8 : (i+1)*8])
		if n < 8 {
			return nil, nil, errors.New("varint read mismatch")
		}
		intIDs[len(intIDs)-1-i] = intID
	}

	var key *datastore.Key
	for i := range intIDs {
		if intIDs[i] == 0 {
			continue
		}
		key = datastore.NewKey(ctx, kinds[i], "", intIDs[i], key)
		if key == nil {
			return nil, nil, errors.New("could not construct key from token")
		}
	}

	return key, b[len(kinds)*8:], nil
}

// Overwrite or re-insert a document, return the new document ID if re-inserted.
func (col *Collection) Update(id int, data []byte) (newID int, err error) {
	if len(data) > DOC_MAX_ROOM {
		return 0, errors.New("Document is too large")
	} else if id < 0 || id >= col.Used-DOC_HEADER || col.Buf[id] != 1 {
		return 0, errors.New("Document does not exist")
	} else if room, _ := binary.Varint(col.Buf[id+1 : id+11]); room > DOC_MAX_ROOM {
		return 0, errors.New("Document does not exist")
	} else if docEnd := id + DOC_HEADER + int(room); docEnd >= col.Size {
		return 0, errors.New("Document does not exist")
	} else if len(data) <= int(room) {
		padding := id + DOC_HEADER + len(data)
		paddingEnd := id + DOC_HEADER + int(room)
		// Overwrite data and then overwrite padding
		copy(col.Buf[id+DOC_HEADER:padding], data)
		for ; padding < paddingEnd; padding += LEN_PADDING {
			copySize := LEN_PADDING
			if padding+LEN_PADDING >= paddingEnd {
				copySize = paddingEnd - padding
			}
			copy(col.Buf[padding:padding+copySize], PADDING)
		}
		return id, nil
	} else {
		// No enough room - re-insert the document
		col.Delete(id)
		return col.Insert(data)
	}
}