// Given a set of bytes and a type, return a protocol buffer object.
func getProto(msgtype int, data []byte) (*riakMessage, error) {
	var ret *riakMessage = nil

	switch msgtype {
	case 0x09:
		obj := &riak.RpbGetReq{}
		err := proto.Unmarshal(data, obj)
		if err != nil {
			return nil, err
		}

		ret = &riakMessage{method: "get", bucket: []byte(obj.Bucket),
			key: []byte(obj.Key)}
	case 0x0a:
		// get response
	case 0x0b:
		obj := &riak.RpbPutReq{}
		err := proto.Unmarshal(data, obj)
		if err != nil {
			return nil, err
		}

		ret = &riakMessage{method: "put", bucket: []byte(obj.Bucket),
			key: []byte(obj.Key)}
	case 0x0c:
		// put response
	}

	return ret, nil
}

//export frameworkMessageCB
func frameworkMessageCB(
	ptr unsafe.Pointer,
	executorIdMessage *C.ProtobufObj,
	slaveIdMessage *C.ProtobufObj,
	dataMessage *C.ProtobufObj) {
	if ptr != nil {
		var driver *SchedulerDriver = (*SchedulerDriver)(ptr)
		if driver.Scheduler.FrameworkMessage == nil {
			return
		}
		executorData := C.GoBytes(executorIdMessage.data, C.int(executorIdMessage.size))
		var executorId ExecutorID
		err := proto.Unmarshal(executorData, &executorId)
		if err != nil {
			return
		}

		slaveData := C.GoBytes(slaveIdMessage.data, C.int(slaveIdMessage.size))
		var slaveId SlaveID
		err = proto.Unmarshal(slaveData, &slaveId)
		if err != nil {
			return
		}

		message := C.GoBytes(dataMessage.data, C.int(dataMessage.size))
		var messageString string = string(message)

		driver.Scheduler.FrameworkMessage(driver, executorId, slaveId, messageString)
	}
}

//export executorLostCB
func executorLostCB(
	ptr unsafe.Pointer,
	executorIdMessage *C.ProtobufObj,
	slaveIdMessage *C.ProtobufObj,
	status C.int) {
	if ptr != nil {
		var driver *SchedulerDriver = (*SchedulerDriver)(ptr)
		if driver.Scheduler.ExecutorLost == nil {
			return
		}

		executorData := C.GoBytes(executorIdMessage.data, C.int(executorIdMessage.size))
		var executorId ExecutorID
		err := proto.Unmarshal(executorData, &executorId)
		if err != nil {
			return
		}

		slaveData := C.GoBytes(slaveIdMessage.data, C.int(slaveIdMessage.size))
		var slaveId SlaveID
		err = proto.Unmarshal(slaveData, &slaveId)
		if err != nil {
			return
		}

		driver.Scheduler.ExecutorLost(driver, executorId, slaveId, int(status))
	}
}

func readReply(conn *net.TCPConn) {
	length := int32(0)
	binary.Read(conn, binary.LittleEndian, &length)
	rbuf := make([]byte, length)
	io.ReadFull(conn, rbuf)
	if len(rbuf) < 9 {
		return
	}
	cmd := int(rbuf[8])
	switch cmd {
	case S2C_HelloInfoResult_CMD:
		res := &Auth_S2C.HelloInfoResult{}
		err := proto.Unmarshal(rbuf[9:], res)
		if err != nil {
			log.Fatal(err)
		}
		log.Print(res.GetToken())
	case S2C_LoginUserInfo_CMD:
		res := &Auth_S2C.LoginUserInfo{}
		err := proto.Unmarshal(rbuf[9:], res)
		if err != nil {
			log.Fatal(err)
		}
		log.Print(res.GetMyInfo(), res.GetServerTime())
	case ERROR_ErrorInfo_CMD:
		res := &Auth_S2C.ErrorInfo{}
		err := proto.Unmarshal(rbuf[9:], res)
		if err != nil {
			log.Fatal(err)
		}
		log.Print("error type: ", res.GetType())
	default:
		fmt.Println("Unexpected CMD: ", cmd)
	}
}

//export registeredCB
func registeredCB(
	ptr unsafe.Pointer,
	frameworkMessage *C.ProtobufObj,
	masterMessage *C.ProtobufObj) {
	if ptr != nil {
		var driver *SchedulerDriver = (*SchedulerDriver)(ptr)

		if driver.Scheduler.Registered == nil {
			return
		}

		frameworkData := C.GoBytes(
			frameworkMessage.data,
			C.int(frameworkMessage.size))

		var frameworkId FrameworkID
		err := proto.Unmarshal(frameworkData, &frameworkId)
		if err != nil {
			return
		}

		masterData := C.GoBytes(masterMessage.data, C.int(masterMessage.size))
		var masterInfo MasterInfo
		err = proto.Unmarshal(masterData, &masterInfo)
		if err != nil {
			return
		}

		driver.Scheduler.Registered(driver, frameworkId, masterInfo)
	}
}

func deserializeByReader(reader *bufio.Reader) (*pb.MobileSuiteProtobuf, error) {
	buff, _ := reader.Peek(4)
	data := bytes.NewBuffer(buff)
	var length int32
	err := binary.Read(data, binary.BigEndian, &length)
	if err != nil {
		return nil, err
	}
	fmt.Println(length)
	if int32(reader.Buffered()) < length+4 {
		return nil, err
	}

	pack := make([]byte, int(4+length))
	_, err = reader.Read(pack)
	if err != nil {
		return nil, err
	}
	msg := pack[4:]
	var dst pb.MobileSuiteProtobuf
	proto.Unmarshal(msg, &dst)
	fmt.Println(&dst)

	var testMsg pb.TestMessage
	proto.Unmarshal(dst.Message, &testMsg)
	fmt.Println(&testMsg)

	return &dst, nil
}

func Marshal() {
	var Test testdata.GoEnum
	Test.Foo = new(testdata.FOO)
	fmt.Println(Test)

	data, err := proto.Marshal(&Test)
	if checkerr(err) {
		return
	}
	fmt.Println(data)

	var goe testdata.GoEnum
	err = proto.Unmarshal(data, &goe)
	if checkerr(err) {
		return
	}

	fmt.Println(goe)

	var i proto.Message
	err = proto.Unmarshal(data, i)
	if checkerr(err) {
		return
	}

	fmt.Println(i)
}

// read a reply to a buffer based on the expected message type
// return error if reply message has different type of command than expected
func readHelloReply(conn *net.TCPConn) (proto.Message, error) {
	duration := time.Second * 10
	timeNow := time.Now()
	err := conn.SetReadDeadline(timeNow.Add(duration))
	if err != nil {
		TimeEncodedPrint("Cant set read timeout", err.Error())
		return nil, err
	}
	length := int32(0)
	err = binary.Read(conn, binary.LittleEndian, &length)
	if DEBUG_READING_MESSAGE {
		fmt.Println("TRYING TO READ MESSAGE LENGTH => ", length, " ERROR: ", err)
	}

	if err != nil {
		return nil, err
	}

	if BYPASS_CONNECTION_SERVER {
		tempBuf := make([]byte, 8)
		_, err = io.ReadFull(conn, tempBuf)
		if DEBUG_READING_MESSAGE {
			fmt.Println("Trying to read extra 8 bytes:", tempBuf, " PROBLEM: ", err)
		}

		if err != nil {
			return nil, err
		}
		length = length - 8
	}

	rbuf := make([]byte, length)
	n, err := io.ReadFull(conn, rbuf)

	cmd := int(rbuf[0])

	switch cmd {
	case S2C_HelloInfoResult_CMD:
		res := &Auth_S2C.HelloInfoResult{}
		err := proto.Unmarshal(rbuf[1:], res)
		if err != nil {
			log.Fatal(err)
		} else {
			return res, nil
		}

	case ERROR_ErrorInfo_CMD:
		res := &Auth_S2C.ErrorInfo{}
		err := proto.Unmarshal(rbuf[1:], res)
		if err != nil {
			log.Fatal(err)
		}
		msg := fmt.Sprint("Server returns error: ")
		return res, errors.New(msg)
	default:
		log.Fatal("Unexpected CMD: ", cmd, "length is: ", length, " number bytes read: ", n, " error: ", err)
	}
	return nil, nil

}

// Deserializes the data from possibly multiple packets,
// currently only for pb.RpbListKeysResp.
func (c *Client) mp_response(conn *net.TCPConn) (response [][]byte, err error) {
	defer c.releaseConn(conn)
	var (
		partial *pb.RpbListKeysResp
		msgcode byte
	)

	for {
		// Read the response from Riak
		msgbuf, err := c.read(conn, 5)
		if err != nil {
			return nil, err
		}
		// Check the length
		if len(msgbuf) < 5 {
			return nil, BadResponseLength
		}
		// Read the message length, read the rest of the message if necessary
		msglen := int(msgbuf[0])<<24 + int(msgbuf[1])<<16 + int(msgbuf[2])<<8 + int(msgbuf[3])
		pbmsg, err := c.read(conn, msglen-1)
		if err != nil {
			return nil, err
		}

		// Deserialize, by default the calling method should provide the expected RbpXXXResp
		msgcode = msgbuf[4]

		if msgcode == messageCodes["RpbListKeysResp"] {
			partial = &pb.RpbListKeysResp{}
			err = proto.Unmarshal(pbmsg, partial)
			if err != nil {
				return nil, err
			}

			response = append(response, partial.Keys...)

			if partial.Done != nil {
				break
			}
		} else if msgcode == messageCodes["RpbErrorResp"] {
			errResp := &pb.RpbErrorResp{}
			err = proto.Unmarshal(pbmsg, errResp)
			if err == nil {
				err = errors.New(string(errResp.Errmsg))
			} else {
				err = fmt.Errorf("Cannot deserialize error response from Riak - %v", err)
			}
			return nil, err
		} else {
			return nil, err
		}
	}

	return
}

// read a reply to a buffer based on the expected message type
// return error if reply message has different type of command than expected
func readReply(conn *net.TCPConn) (proto.Message, error) {
	duration := time.Millisecond * 20
	timeNow := time.Now()
	err := conn.SetReadDeadline(timeNow.Add(duration))
	if err != nil {
		TimeEncodedPrint("Cant set read timeout", err.Error())
		return nil, err
	}
	length := int32(0)
	err = binary.Read(conn, binary.LittleEndian, &length)
	if err != nil {
		return nil, nil
	}

	if DEBUG_STEP {
		fmt.Println("RECEIVE MESSAGE LENGTH: ", length)
	}
	// now wait longer for the message data
	duration = time.Millisecond * 100
	timeNow = time.Now()
	err = conn.SetReadDeadline(timeNow.Add(duration))

	rbuf := make([]byte, length)
	io.ReadFull(conn, rbuf)
	if DEBUG_STEP {
		fmt.Println(rbuf)
	}

	cmd := int(rbuf[0])

	switch cmd {
	case S2C_HelloInfoResult_CMD:
		res := &Auth_S2C.HelloInfoResult{}
		err := proto.Unmarshal(rbuf[1:], res)
		if err != nil {
			log.Fatal(err)
		} else {
			return res, nil
		}

	case ERROR_ErrorInfo_CMD:
		res := &Auth_S2C.ErrorInfo{}
		err := proto.Unmarshal(rbuf[1:], res)
		if err != nil {
			log.Fatal(err)
		}
		msg := fmt.Sprint("Server returns error: ")
		return res, errors.New(msg)
	default:
		log.Fatal("Unexpected CMD: ", cmd)
	}
	return nil, nil

}

func (contact *Contact) processKeyExchange(kxsBytes []byte, testing bool) error {
	var kxs pond.SignedKeyExchange
	if err := proto.Unmarshal(kxsBytes, &kxs); err != nil {
		return err
	}

	var sig [64]byte
	if len(kxs.Signature) != len(sig) {
		return errors.New("invalid signature length")
	}
	copy(sig[:], kxs.Signature)

	var kx pond.KeyExchange
	if err := proto.Unmarshal(kxs.Signed, &kx); err != nil {
		return err
	}

	if len(kx.PublicKey) != len(contact.theirPub) {
		return errors.New("invalid public key")
	}
	copy(contact.theirPub[:], kx.PublicKey)

	if !ed25519.Verify(&contact.theirPub, kxs.Signed, &sig) {
		return errors.New("invalid signature")
	}

	contact.theirServer = *kx.Server
	if _, _, err := parseServer(contact.theirServer, testing); err != nil {
		return err
	}

	group, ok := new(bbssig.Group).Unmarshal(kx.Group)
	if !ok {
		return errors.New("invalid group")
	}
	if contact.myGroupKey, ok = new(bbssig.MemberKey).Unmarshal(group, kx.GroupKey); !ok {
		return errors.New("invalid group key")
	}

	if len(kx.IdentityPublic) != len(contact.theirIdentityPublic) {
		return errors.New("invalid public identity")
	}
	copy(contact.theirIdentityPublic[:], kx.IdentityPublic)

	if len(kx.Dh) != len(contact.theirCurrentDHPublic) {
		return errors.New("invalid public DH value")
	}
	copy(contact.theirCurrentDHPublic[:], kx.Dh)

	contact.generation = *kx.Generation

	return nil
}

func Recv(conn net.Conn) {
	defer func() { sendChan <- nil }()

	for {
		nethead := [HeadLen]byte{}
		if _, e := io.ReadFull(conn, nethead[:]); e != nil {
			fmt.Println(conn.RemoteAddr().String(), "Error recv head : ", e)
			break
		}
		msglen := *(*uint64)(unsafe.Pointer(&nethead))
		msgbuf := make([]byte, HeadLen+msglen)

		if _, e := io.ReadFull(conn, msgbuf[HeadLen:]); e != nil {
			fmt.Println(conn.RemoteAddr().String(), "Error recv body : ", e)
			break
		}

		mh := (*MessageHead)(unsafe.Pointer(&msgbuf[0]))
		mh.Msglen = msglen
		if uint32(len(msgbuf)) < MsgLen+RecLen {
			continue
		}

		rh := (*RecordHead)(unsafe.Pointer(&msgbuf[MsgLen]))

		if uint32(len(msgbuf)) < MsgLen+RecLen+rh.Length {
			continue
		}

		bodybuf := msgbuf[MsgLen+RecLen : MsgLen+RecLen+rh.Length]
		if rh.HashCode != Common.Hash(bodybuf) {
			continue
		}

		switch rh.RecType {
		case uint32(ConstDefine_E_DepthMarketData):
			dmd := &DepthMarketData{}
			proto.Unmarshal(bodybuf, dmd)
			quoteChan <- dmd

		case uint32(ConstDefine_E_ReqKLine):
			rkl := &ReqKLine{}
			proto.Unmarshal(bodybuf, rkl)
			klineChan <- &KlineReq{mh.Sender, mh.Copyer, mh.SendTime, rkl}

		case uint32(ConstDefine_E_ReqTick):
			rt := &ReqTick{}
			proto.Unmarshal(bodybuf, rt)
			tickChan <- &TickReq{mh.Sender, mh.Copyer, mh.SendTime, rt}
		}
	}
}

func availabilityIntervals(inputChan, outputChan chan *store.Record) {
	writeRecord := func(firstKey, lastKey *TraceKey, firstTrace, lastTrace []byte) {
		firstTraceDecoded := Trace{}
		if err := proto.Unmarshal(firstTrace, &firstTraceDecoded); err != nil {
			log.Fatalf("Error ummarshaling protocol buffer: %v", err)
		}
		lastTraceDecoded := Trace{}
		if err := proto.Unmarshal(lastTrace, &lastTraceDecoded); err != nil {
			log.Fatalf("Error ummarshaling protocol buffer: %v", err)
		}
		intervalKey := intervalKey{
			NodeId:               firstKey.NodeId,
			AnonymizationContext: firstKey.AnonymizationContext,
			SessionId:            firstKey.SessionId,
			FirstSequenceNumber:  firstKey.SequenceNumber,
			LastSequenceNumber:   lastKey.SequenceNumber,
		}
		outputChan <- &store.Record{
			Key:   encodeIntervalKey(&intervalKey),
			Value: lex.EncodeOrDie(*firstTraceDecoded.TraceCreationTimestamp, *lastTraceDecoded.TraceCreationTimestamp),
		}
	}

	var firstKey, lastKey *TraceKey
	var firstTrace, lastTrace []byte
	var expectedTraceKey []byte
	for record := range inputChan {
		var traceKey TraceKey
		lex.DecodeOrDie(record.Key, &traceKey)
		expectedNextTraceKeyDecoded := TraceKey{
			NodeId:               traceKey.NodeId,
			AnonymizationContext: traceKey.AnonymizationContext,
			SessionId:            traceKey.SessionId,
			SequenceNumber:       traceKey.SequenceNumber + 1,
		}
		expectedNextTraceKey := lex.EncodeOrDie(&expectedNextTraceKeyDecoded)

		if !bytes.Equal(expectedTraceKey, record.Key) {
			if firstTrace != nil {
				writeRecord(firstKey, lastKey, firstTrace, lastTrace)
			}
			firstKey = &traceKey
			firstTrace = record.Value
		}
		lastKey = &traceKey
		lastTrace = record.Value
		expectedTraceKey = expectedNextTraceKey
	}
	if firstTrace != nil {
		writeRecord(firstKey, lastKey, firstTrace, lastTrace)
	}
}

func (c *context) Call(service, method string, in, out proto.Message, opts *internal.CallOptions) error {
	req, err := proto.Marshal(in)
	if err != nil {
		return fmt.Errorf("error marshalling request: %v", err)
	}

	remReq := &pb.Request{
		ServiceName: proto.String(service),
		Method:      proto.String(method),
		Request:     req,
		// NOTE(djd): RequestId is unused in the server.
	}

	req, err = proto.Marshal(remReq)
	if err != nil {
		return fmt.Errorf("proto.Marshal: %v", err)
	}

	// TODO(djd): Respect opts.Timeout?
	resp, err := c.client.Post(c.url, "application/octet-stream", bytes.NewReader(req))
	if err != nil {
		return fmt.Errorf("error sending request: %v", err)
	}
	defer resp.Body.Close()

	body, err := ioutil.ReadAll(resp.Body)
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("bad response %d; body: %q", resp.StatusCode, body)
	}
	if err != nil {
		return fmt.Errorf("failed reading response: %v", err)
	}
	remResp := &pb.Response{}
	if err := proto.Unmarshal(body, remResp); err != nil {
		return fmt.Errorf("error unmarshalling response: %v", err)
	}

	if ae := remResp.GetApplicationError(); ae != nil {
		return &internal.APIError{
			Code:    ae.GetCode(),
			Detail:  ae.GetDetail(),
			Service: service,
		}
	}

	if remResp.Response == nil {
		return fmt.Errorf("unexpected response: %s", proto.MarshalTextString(remResp))
	}

	return proto.Unmarshal(remResp.Response, out)
}

func tryIgnoreReceivedMessage(buffer []byte) bool {
	if DEBUG_IGNORING_MESSAGE {
		fmt.Println("Trying to ignore received message buffer: ", buffer)
		fmt.Println("Global ignore list:", globalIgnoreMessages)
		fmt.Println("Current ignore list:", currentIgnoreMessages)
	}

	baseCmd := buffer[0]
	// report error
	signature := calculateSignature(0, int(baseCmd))
	if DEBUG_IGNORING_MESSAGE {
		fmt.Println("Signature 1: ", signature)
	}
	if ignoreType, exist := currentIgnoreMessages[signature]; exist {
		newValue := magicVarFunc(ignoreType)
		res := newValue.(proto.Message)
		err := proto.Unmarshal(buffer[1:], res)
		if err == nil {
			// can ignore this message
			if DEBUG_IGNORING_MESSAGE {
				fmt.Println("Ignored!")
			}
			return true
		}
	}

	if len(buffer) >= 2 {
		signature = calculateSignature(int(baseCmd), int(buffer[1]))
		if DEBUG_IGNORING_MESSAGE {
			fmt.Println("Signature 2: ", signature)
		}
		if ignoreType, exist := currentIgnoreMessages[signature]; exist {
			newValue := magicVarFunc(ignoreType)
			res := newValue.(proto.Message)
			err := proto.Unmarshal(buffer[1:], res)
			if err == nil {
				// can ignore this message
				if DEBUG_IGNORING_MESSAGE {
					fmt.Println("Ignored!")
				}
				return true
			}
		}
	}
	if DEBUG_IGNORING_MESSAGE {
		fmt.Println("Not ignored!")
	}
	return false
}

func (c *context) Call(service, method string, in, out ProtoMessage, opts *CallOptions) error {
	if service == "__go__" {
		if method == "GetNamespace" {
			out.(*basepb.StringProto).Value = proto.String(c.req.Header.Get("X-AppEngine-Current-Namespace"))
			return nil
		}
		if method == "GetDefaultNamespace" {
			out.(*basepb.StringProto).Value = proto.String(c.req.Header.Get("X-AppEngine-Default-Namespace"))
			return nil
		}
	}
	if f, ok := apiOverrides[struct{ service, method string }{service, method}]; ok {
		return f(in, out, opts)
	}
	data, err := proto.Marshal(in)
	if err != nil {
		return err
	}

	requestID := c.req.Header.Get("X-Appengine-Internal-Request-Id")
	res, err := call(service, method, data, requestID)
	if err != nil {
		return err
	}
	return proto.Unmarshal(res, out)
}

func (c *Connection) ReadResponse(s *Session, token int64) (*p.Response, error) {
	for {
		var messageLength uint32
		if err := binary.Read(c, binary.LittleEndian, &messageLength); err != nil {
			c.Close()
			return nil, RqlConnectionError{err.Error()}
		}

		buffer := make([]byte, messageLength)
		if _, err := io.ReadFull(c, buffer); err != nil {
			c.Close()
			return nil, RqlDriverError{err.Error()}
		}

		response := &p.Response{}
		if err := proto.Unmarshal(buffer, response); err != nil {
			return nil, RqlDriverError{err.Error()}
		}

		if response.GetToken() == token {
			return response, nil
		} else if cursor, ok := s.checkCache(token); ok {
			// Handle batch response
			s.handleBatchResponse(cursor, response)
		} else {
			return nil, RqlDriverError{"Unexpected response received"}
		}
	}
}

func (conn *ProtoBufConn) GetRequestBody(req *protobuf.Packet, body interface{}) error {
	if value, ok := body.(proto.Message); ok {
		return proto.Unmarshal(req.GetSerializedPacket(), value)
	}

	return fmt.Errorf("GetRequestBody value type error %v", body)
}

func (conn *ProtoBufConn) ReadRequest(req *protobuf.Packet) error {

	conn.c.SetReadDeadline(time.Now().Add(ConnReadTimeOut))

	dst, err := conn.c.ReadMessage()
	if err != nil {
		logger.Debug("ReadRequest Read binary Err: %v", err)
		return err
	}

	//dst, err := snappy.Decode(nil, dstBuffer.Bytes())

	if err != nil {
		logger.Debug("ReadRequest Decode Err: %v", err)
		return err
	}

	conn.last_time = time.Now().Unix()

	//logger.Info("ReadRequest dst: %v", dst)

	err = proto.Unmarshal(dst, req)
	conn.msg_id = req.GetId()
	return err
}

func Consume(connection *amqp.Connection, queue string, outChannel chan *protobuf.Notification) {
	// create a channel on this connection
	channel, err := connection.Channel()
	if err != nil {
		panic(err)
	}
	defer channel.Close()

	// start consuming data
	consumerTag := queue + "-consumer"
	deliveries, err := channel.Consume(
		queue,       // name
		consumerTag, // consumerTag
		false,       // noAck
		false,       // exclusive
		false,       // noLocal
		false,       // noWait
		nil,         // arguments
	)
	if err != nil {
		panic(err)
	}
	defer channel.Cancel(consumerTag, false)

	for delivery := range deliveries {
		notif := &protobuf.Notification{}
		proto.Unmarshal(delivery.Body, notif)

		outChannel <- notif

		if err := delivery.Ack(false); err != nil {
			panic(err)
		}
	}
}