// CreateFile opens a new file in HDFS with the given replication, block size,
// and permissions, and returns an io.WriteCloser for writing to it. Because of
// the way that HDFS writes are buffered and acknowledged asynchronously, it is
// very important that Close is called after all data has been written.
func (c *Client) CreateFile(name string, replication int, blockSize int64, perm os.FileMode) (*FileWriter, error) {
	createReq := &hdfs.CreateRequestProto{
		Src:          proto.String(name),
		Masked:       &hdfs.FsPermissionProto{Perm: proto.Uint32(uint32(perm))},
		ClientName:   proto.String(c.namenode.ClientName()),
		CreateFlag:   proto.Uint32(1),
		CreateParent: proto.Bool(false),
		Replication:  proto.Uint32(uint32(replication)),
		BlockSize:    proto.Uint64(uint64(blockSize)),
	}
	createResp := &hdfs.CreateResponseProto{}

	err := c.namenode.Execute("create", createReq, createResp)
	if err != nil {
		if nnErr, ok := err.(*rpc.NamenodeError); ok {
			err = interpretException(nnErr.Exception, err)
		}

		return nil, &os.PathError{"create", name, err}
	}

	return &FileWriter{
		client:      c,
		name:        name,
		replication: replication,
		blockSize:   blockSize,
	}, nil
}

func createBlock(t *testing.T, name string) *BlockWriter {
	namenode := getNamenode(t)
	blockSize := int64(1048576)

	createReq := &hdfs.CreateRequestProto{
		Src:          proto.String(name),
		Masked:       &hdfs.FsPermissionProto{Perm: proto.Uint32(uint32(0644))},
		ClientName:   proto.String(namenode.ClientName()),
		CreateFlag:   proto.Uint32(1),
		CreateParent: proto.Bool(false),
		Replication:  proto.Uint32(uint32(3)),
		BlockSize:    proto.Uint64(uint64(blockSize)),
	}
	createResp := &hdfs.CreateResponseProto{}

	err := namenode.Execute("create", createReq, createResp)
	require.NoError(t, err)

	addBlockReq := &hdfs.AddBlockRequestProto{
		Src:        proto.String(name),
		ClientName: proto.String(namenode.ClientName()),
		Previous:   nil,
	}
	addBlockResp := &hdfs.AddBlockResponseProto{}

	err = namenode.Execute("addBlock", addBlockReq, addBlockResp)
	require.NoError(t, err)

	block := addBlockResp.GetBlock()
	return NewBlockWriter(block, namenode, blockSize)
}

// Update the datastore with the user's current state.
func (server *Server) UpdateFrozenUser(client *Client, state *mumbleproto.UserState) {
	// Full sync If there's no userstate messgae provided, or if there is one, and
	// it includes a registration operation.
	user := client.user
	nanos := time.Now().Unix()
	if state == nil || state.UserId != nil {
		fu, err := user.Freeze()
		if err != nil {
			server.Fatal(err)
		}
		fu.LastActive = proto.Uint64(uint64(nanos))
		err = server.freezelog.Put(fu)
		if err != nil {
			server.Fatal(err)
		}
	} else {
		fu := &freezer.User{}
		fu.Id = proto.Uint32(user.Id)
		if state.ChannelId != nil {
			fu.LastChannelId = proto.Uint32(uint32(client.Channel.Id))
		}
		if state.TextureHash != nil {
			fu.TextureBlob = proto.String(user.TextureBlob)
		}
		if state.CommentHash != nil {
			fu.CommentBlob = proto.String(user.CommentBlob)
		}
		fu.LastActive = proto.Uint64(uint64(nanos))
		err := server.freezelog.Put(fu)
		if err != nil {
			server.Fatal(err)
		}
	}
	server.numLogOps += 1
}

func newPbLog(record *pb.Log) {
	record.Timestamp = proto.Int64(time.Now().UnixNano())
	record.ZoneId = proto.Uint32(123456)
	record.ZonePlan = pb.ZonePlan_FREE.Enum()

	record.Http = &pb.HTTP{
		Protocol:    pb.HTTP_HTTP11.Enum(),
		Status:      proto.Uint32(200),
		HostStatus:  proto.Uint32(503),
		UpStatus:    proto.Uint32(520),
		Method:      pb.HTTP_GET.Enum(),
		ContentType: proto.String("text/html"),
		UserAgent:   proto.String(userAgent),
		Referer:     proto.String("https://www.cloudflare.com/"),
		RequestURI:  proto.String("/cdn-cgi/trace"),
	}

	record.Origin = &pb.Origin{
		Ip:       []byte(net.IPv4(1, 2, 3, 4).To4()),
		Port:     proto.Uint32(8080),
		Hostname: proto.String("www.example.com"),
		Protocol: pb.Origin_HTTPS.Enum(),
	}

	record.Country = pb.Country_US.Enum()
	record.CacheStatus = pb.CacheStatus_HIT.Enum()
	record.ServerIp = []byte(net.IPv4(192, 168, 1, 1).To4())
	record.ServerName = proto.String("metal.cloudflare.com")
	record.RemoteIp = []byte(net.IPv4(10, 1, 2, 3).To4())
	record.BytesDlv = proto.Uint64(123456)
	record.RayId = proto.String("10c73629cce30078-LAX")
}

func (bw *BlockWriter) writeBlockWriteRequest(w io.Writer) error {
	targets := bw.currentPipeline()[1:]

	op := &hdfs.OpWriteBlockProto{
		Header: &hdfs.ClientOperationHeaderProto{
			BaseHeader: &hdfs.BaseHeaderProto{
				Block: bw.block.GetB(),
				Token: bw.block.GetBlockToken(),
			},
			ClientName: proto.String(bw.clientName),
		},
		Targets:               targets,
		Stage:                 bw.currentStage().Enum(),
		PipelineSize:          proto.Uint32(uint32(len(targets))),
		MinBytesRcvd:          proto.Uint64(bw.block.GetB().GetNumBytes()),
		MaxBytesRcvd:          proto.Uint64(uint64(bw.offset)), // I don't understand these two fields
		LatestGenerationStamp: proto.Uint64(uint64(bw.generationTimestamp())),
		RequestedChecksum: &hdfs.ChecksumProto{
			Type:             hdfs.ChecksumTypeProto_CHECKSUM_CRC32.Enum(),
			BytesPerChecksum: proto.Uint32(outboundChunkSize),
		},
	}

	return writeBlockOpRequest(w, writeBlockOp, op)
}

func (s *Session) HandlePacket(header *rpc.Header, body []byte) {
	if s.state == StateDisconnected {
		panic("cannot handle packets from disconnected clients")
	}
	serviceId := int(header.GetServiceId())
	methodId := int(header.GetMethodId())
	s.receivedToken = header.GetToken()

	if serviceId == 254 {
		s.HandleResponse(header.GetToken(), body)
	} else {
		resp := s.HandleRequest(serviceId, methodId, body)
		if resp != nil {
			respHead := &rpc.Header{
				ServiceId: proto.Uint32(254),
				Token:     header.Token,
				Size:      proto.Uint32(uint32(len(resp))),
			}
			err := s.QueuePacket(respHead, resp)
			if err != nil {
				log.Panicf("error: Session.HandlePacket: respond: %v", err)
			}
		}
	}
}

func (this *SClientUser) send_CmdServer2Gate_GetClientMsgRangeRes() error {
	sendMsg := &bsn_msg_gate_server.SGetClientMsgRangeRes{
		Vu32MsgTypeMin: proto.Uint32(uint32(bsn_msg_client_echo_server.ECmdClient2EchoServer_CmdClient2EchoServer_Min)),
		Vu32MsgTypeMax: proto.Uint32(uint32(bsn_msg_client_echo_server.ECmdClient2EchoServer_CmdClient2EchoServer_Max)),
	}
	return this.SendPbMsgWithSMsgHeader(bsn_common.TMsgType(bsn_msg_gate_server.ECmdServe2Gate_CmdServer2Gate_GetClientMsgRangeRes), sendMsg)
}

func chat(agent *Agent, chatType uint32, targetId uint32, content string) {
	req := &pb.MQChat{}
	req.ChatType = proto.Uint32(chatType)
	req.TargetId = proto.Uint32(targetId)
	req.Content = proto.String(content)
	quest(agent, pb.MCHAT, req)
}

func listener(c <-chan []byte, dst *string) {
	for {
		msg := <-c
		var conAuxSlice []ConAux
		json.Unmarshal(msg, &conAuxSlice)

		fmt.Println("unmarshalled", conAuxSlice)

		connections := new(protobee.Connections)
		connections.Connection = []*protobee.Connection{}

		for _, value := range conAuxSlice {
			con := new(protobee.Connection)
			con.Transport = proto.String(value.Transport)
			con.LocalAddress = proto.String(value.LocalAddress)
			con.LocalPort = proto.Uint32(value.LocalPort)
			con.RemoteAddress = proto.String(value.RemoteAddress)
			con.RemotePort = proto.Uint32(value.RemotePort)
			con.Pid = proto.Uint32(value.Pid)
			con.Name = proto.String(value.Name)
			connections.Connection = append(connections.Connection, con)
		}
		//connections
		pb, err := proto.Marshal(connections)
		if err != nil {
			fmt.Println("error", err)
		}
		sendDataToDest(pb, dst)
		//time.Sleep(time.Second * 2)
	}
}

func (s *AuthServerService) CompleteLogin() error {
	res := authentication_service.LogonResult{}
	if !s.loggedIn {
		res.ErrorCode = proto.Uint32(ErrorNoAuth)
	} else {
		res.ErrorCode = proto.Uint32(ErrorOK)
		// TODO: Make this data real.  ConnectGameServer needs to return the
		// GameAccount EntityId.
		res.Account = EntityId(72058118023938048, 1)
		res.GameAccount = make([]*entity.EntityId, 1)
		res.GameAccount[0] = EntityId(144115713527006023, 1)
		res.ConnectedRegion = proto.Uint32(0x5553) // 'US'

		if s.program == "WTCG" {
			s.sess.server.ConnectGameServer(s.sess, s.program)
			go s.sess.HandleNotifications()
		}
		s.sess.startedPlaying = time.Now()
	}
	resBody, err := proto.Marshal(&res)
	if err != nil {
		return err
	}
	resHeader := s.sess.MakeRequestHeader(s.client, 5, len(resBody))
	err = s.sess.QueuePacket(resHeader, resBody)
	if err != nil {
		return err
	}
	return nil
}

// Load given list of tracks on spotify connect device with given
// ident.  Gids are formated base62 spotify ids.
func (c *SpircController) LoadTrack(ident string, gids []string) error {
	c.seqNr += 1

	tracks := make([]*Spotify.TrackRef, 0, len(gids))
	for _, g := range gids {
		tracks = append(tracks, &Spotify.TrackRef{
			Gid:    convert62(g),
			Queued: proto.Bool(false),
		})
	}

	state := &Spotify.State{
		Index:             proto.Uint32(0),
		Track:             tracks,
		Status:            Spotify.PlayStatus_kPlayStatusStop.Enum(),
		PlayingTrackIndex: proto.Uint32(0),
	}

	frame := &Spotify.Frame{
		Version:         proto.Uint32(1),
		Ident:           proto.String(c.ident),
		ProtocolVersion: proto.String("2.0.0"),
		SeqNr:           proto.Uint32(c.seqNr),
		Typ:             Spotify.MessageType_kMessageTypeLoad.Enum(),
		Recipient:       []string{ident},
		State:           state,
	}

	return c.sendFrame(frame)
}

// Range scan index between low and high.
func (c *GsiScanClient) Range(
	defnID uint64, low, high common.SecondaryKey, inclusion Inclusion,
	distinct bool, limit int64, cons common.Consistency, vector *TsConsistency,
	callb ResponseHandler) (error, bool) {

	// serialize low and high values.
	l, err := json.Marshal(low)
	if err != nil {
		return err, false
	}
	h, err := json.Marshal(high)
	if err != nil {
		return err, false
	}

	connectn, err := c.pool.Get()
	if err != nil {
		return err, false
	}
	healthy := true
	defer func() { c.pool.Return(connectn, healthy) }()

	conn, pkt := connectn.conn, connectn.pkt

	req := &protobuf.ScanRequest{
		DefnID: proto.Uint64(defnID),
		Span: &protobuf.Span{
			Range: &protobuf.Range{
				Low: l, High: h, Inclusion: proto.Uint32(uint32(inclusion)),
			},
		},
		Distinct: proto.Bool(distinct),
		Limit:    proto.Int64(limit),
		Cons:     proto.Uint32(uint32(cons)),
	}
	if vector != nil {
		req.Vector = protobuf.NewTsConsistency(
			vector.Vbnos, vector.Seqnos, vector.Vbuuids, vector.Crc64)
	}
	// ---> protobuf.ScanRequest
	if err := c.sendRequest(conn, pkt, req); err != nil {
		fmsg := "%v Range() request transport failed `%v`\n"
		logging.Errorf(fmsg, c.logPrefix, err)
		healthy = false
		return err, false
	}

	cont, partial := true, false
	for cont {
		// <--- protobuf.ResponseStream
		cont, healthy, err = c.streamResponse(conn, pkt, callb)
		if err != nil { // if err, cont should have been set to false
			fmsg := "%v Range() response failed `%v`\n"
			logging.Errorf(fmsg, c.logPrefix, err)
		} else { // partial succeeded
			partial = true
		}
	}
	return err, partial
}

func NewStartRequest(path string, dir string, args []string, allocated resource.ComputeResource, envs []string, host string, port int32) *cmd.ControlMessage {
	request := &cmd.ControlMessage{
		Type: cmd.ControlMessage_StartRequest.Enum(),
		StartRequest: &cmd.StartRequest{
			Path: proto.String(path),
			Args: args,
			Dir:  proto.String(dir),
			Resource: &cmd.ComputeResource{
				CpuCount: proto.Int32(int32(allocated.CPUCount)),
				CpuLevel: proto.Int32(int32(allocated.CPULevel)),
				Memory:   proto.Int32(int32(allocated.MemoryMB)),
			},
			Envs:     envs,
			Host:     proto.String(host),
			Port:     proto.Int32(port),
			HashCode: proto.Uint32(0),
		},
	}

	// generate a unique hash code for the request
	data, err := proto.Marshal(request)
	if err != nil {
		log.Fatalf("marshaling start request error: %v", err)
		return nil
	}
	request.StartRequest.HashCode = proto.Uint32(uint32(util.Hash(data)))

	return request
}

func (s *AuthServerService) CompleteLogin() error {
	res := hsproto.BnetProtocolAuthentication_LogonResult{}
	if !s.loggedIn {
		res.ErrorCode = proto.Uint32(ErrorNoAuth)
	} else {
		res.ErrorCode = proto.Uint32(ErrorOK)
		res.Account = EntityId(0, 1)
		res.GameAccount = make([]*hsproto.BnetProtocol_EntityId, 1)
		res.GameAccount[0] = EntityId(1, 1)
		res.ConnectedRegion = proto.Uint32(0x5553) // 'US'

		if s.program == "WTCG" {
			s.sess.game = s.sess.server.ConnectGameServer(s.sess, s.program)
		}
		s.sess.startedPlaying = time.Now()
	}
	resBody, err := proto.Marshal(&res)
	if err != nil {
		return err
	}
	resHeader := s.sess.MakeRequestHeader(s.client, 5, len(resBody))
	err = s.sess.QueuePacket(resHeader, resBody)
	if err != nil {
		return err
	}
	return nil
}

func toRoleBasic(r *RoleBasic) *pb.RoleBasic {
	b := &pb.RoleBasic{}
	b.Id = proto.Uint32(r.id)
	b.Occupation = proto.Uint32(r.occupation)
	b.Level = proto.Uint32(r.level)
	b.Name = proto.String(r.name)
	return b
}

func (f *ConcreteMsgFactory) CreateFollowerInfo(epoch uint32,
	fid string,
	voting bool) protocol.FollowerInfoMsg {

	return &FollowerInfo{Version: proto.Uint32(ProtoVersion()),
		AcceptedEpoch: proto.Uint32(epoch),
		Fid:           proto.String(fid),
		Voting:        proto.Bool(voting)}
}

//登录SuperService
func (service *superRpc) Login(req *msg.LoginRequst, res *msg.LoginRespose) error {
	log.Trace("[SUPER] %s(%s) 的连接到来", utils.GetServiceName(int(req.GetServiceType())), string(req.GetServiceIp()))
	res.ServiceType = proto.Uint32(req.GetServiceType())
	res.ServiceId = proto.Uint64(100)
	res.ServiceIp = proto.String(req.GetServiceIp())
	res.RetCode = proto.Uint32(1)
	res.ExterPort = proto.Uint32(300)
	log.Trace("[SUPER] 返回 %s 服务器ID %d", utils.GetServiceName(int(res.GetServiceType())), res.GetServiceId())
	return nil
}

func (s *Session) Respond(token uint32, body []byte) {
	err := s.QueuePacket(&rpc.Header{
		ServiceId: proto.Uint32(254),
		Token:     proto.Uint32(token),
		Size:      proto.Uint32(uint32(len(body))),
	}, body)
	if err != nil {
		log.Panicf("error: Session.Respond: %v", err)
	}
}

func main() {
	fmt.Println("XMAN Test")

	// argv: name default_val introduction
	configFile := flag.String("conf", "./config/srv.ini", "server config file.")
	flag.Parse()

	fmt.Println(*configFile)
	config := InitConfig(*configFile)
	section, err := config.Section("MysqlInfo")
	if err != nil {
		fmt.Println(err)
		return
	}
	fmt.Println(section.ValueOf("host"))

	test := &xmandemo.BaseHeader{
		Ver: proto.Uint32(111),
		Bus: proto.Uint32(222),
		Seq: proto.Uint32(1),
		Cmd: proto.Uint32(uint32(xmandemo.XCMD_CMD_DEMO)),
	}
	data, err := proto.Marshal(test)
	fmt.Println(data)

	//	utils.Daemonize(0, 1)

	SetConsoleShow(true)
	//	SetRollingDaily("/Users/reezhou/Desktop/xman/src/logs", "test.log")
	SetRollingFile("/Users/reezhou/Desktop/xman/src/logs", "rolling.log", 10, 50, MB)
	SetLogLevel(LOG_DEBUG)
	//	Log(LOG_ERROR, "uin error")

	// redis test
	//	redisController = xmandb.NewRedisController()
	//	defer redisController.Close()
	//	err = redisController.InitRedis("", ":6379", 0)
	//	if err != nil {
	//		fmt.Println("redis init error.",err)
	//	}
	//	v, _ := redis.String(redisController.Get("ree"), err)
	//	fmt.Println("get from redis: ", v)
	//
	//	// mysql test
	//	mysqlController = xmandb.NewMysqlController()
	//	err = mysqlController.InitMysql(`{"user": "root", "password": "111", "address": ":3306", "dbname": "xman"}`)
	//	defer mysqlController.Close()
	//	if err != nil {
	//		fmt.Println(err)
	//	}
	//	row, _ := mysqlController.FetchRows("SELECT * FROM user")
	//	fmt.Println(row)

	goServer(config)
}

func GetRoleListCallback(conn *GxTcpConn, info *LoginInfo, msg *GxMessage) {
	rdClient := PopRedisClient()
	defer PushRedisClient(rdClient)

	var req GetRoleListReq
	var rsp GetRoleListRsp
	err := msg.UnpackagePbmsg(&req)
	if err != nil {
		SendPbMessage(conn, false, msg.GetId(), msg.GetCmd(), msg.GetSeq(), RetFail, nil)
		return
	}
	if req.Info == nil || req.GetInfo().Token == nil || req.ServerId == nil {
		SendPbMessage(conn, false, msg.GetId(), msg.GetCmd(), msg.GetSeq(), RetMsgFormatError, nil)
		return
	}

	playerName := CheckToken(rdClient, req.GetInfo().GetToken())
	if playerName == "" {
		SendPbMessage(conn, false, msg.GetId(), msg.GetCmd(), msg.GetSeq(), RetTokenError, nil)
		return
	}

	player := new(Player)
	err = player.Get(rdClient, playerName)
	if err != nil {
		SendPbMessage(conn, false, msg.GetId(), msg.GetCmd(), msg.GetSeq(), RetUserNotExists, nil)
		return
	}

	ids := GetRoleList(rdClient, playerName, req.GetServerId())
	for i := 0; i < len(ids); i++ {
		id, _ := strconv.Atoi(ids[i])

		role := new(Role)
		err = role.Get(rdClient, uint32(id))
		if err != nil {
			Debug("role %d is not existst", id)
			continue
		}
		rsp.Roles = append(rsp.Roles, &RoleCommonInfo{
			Id:         proto.Uint32(role.Id),
			Name:       proto.String(role.Name),
			Level:      proto.Uint32(role.Level),
			VocationId: proto.Uint32(role.VocationId),
		})
	}

	info.PlayerName = playerName
	info.BeginTs = time.Now().Unix()
	info.ServerId = req.GetServerId()
	SaveGateLoginInfo(rdClient, info.GateId, info.ConnId, playerName)
	info.Save(rdClient)

	SendPbMessage(conn, false, msg.GetId(), msg.GetCmd(), msg.GetSeq(), RetSucc, &rsp)
}