// 客户端申请加入讨论组
func HandleClientJoinGroup(conn *net.TCPConn, recPacket *packet.Packet) {
	// read
	readMsg := &pb.PbClientJoinGroup{}
	packet.Unpack(recPacket, readMsg)

	from_uuid := readMsg.GetFromUuid()
	group_id := readMsg.GetGroupId()
	// note_msg := readMsg.GetNoteMsg()
	// timestamp := readMsg.GetTimestamp()

	// 加入讨论组
	if ret := groupinfo.JoinGroup(from_uuid, group_id); !ret {
		return
	}

	group_name, _ := groupinfo.GetGroupNameAndOwner(group_id)

	// write
	writeMsg := pb.PbServerNotifyJoinGroup{
		ApplicantUuid: proto.String(from_uuid),
		GroupId:       proto.String(group_id),
		GroupName:     proto.String(group_name),
		Timestamp:     proto.Int64(time.Now().Unix()),
	}

	// 通知所有组员，这个消息不离线存储
	group_members := groupinfo.GetAllUuid(group_id)
	for i := 1; i < len(group_members); i++ {
		SendPbData(UuidMapConn.Get(group_members[i]).(*net.TCPConn), packet.PK_ServerNotifyJoinGroup, writeMsg)
	}
}

func TestFrameworkMessageHandling(t *testing.T) {
	sched := NewMesosScheduler()
	sched.FrameworkMessage = func(schedDriver *SchedulerDriver, execId *mesos.ExecutorID, slaveId *mesos.SlaveID, data []byte) {
		if execId.GetValue() != "test-executor-1" {
			log.Fatal("Scheduler.FrameworkMessage.ExecutorId not received.")
		}

		if slaveId.GetValue() != "test-slave-1" {
			log.Fatal("Scheduler.FrameworkMessage.SlaveId not received.")
		}

		if string(data) != "Hello-Test" {
			log.Fatal("Scheduler.FrameworkMessage.Data not received.")
		}
	}

	msg := &mesos.ExecutorToFrameworkMessage{
		SlaveId:     &mesos.SlaveID{Value: proto.String("test-slave-1")},
		FrameworkId: &mesos.FrameworkID{Value: proto.String("test-framework-1")},
		ExecutorId:  &mesos.ExecutorID{Value: proto.String("test-executor-1")},
		Data:        []byte("Hello-Test"),
	}
	driver, err := NewSchedDriver(sched, &mesos.FrameworkInfo{}, "localhost:0")
	if err != nil {
		t.Fatal(err)
	}
	driver.schedMsgQ <- msg

}

// TODO: only download new resources if updated versions
func netGetClResources(e *Engine) netHandler {
	return func(msg messages.GetCLResources, conn netengine.Connection) []byte {
		appendResourceScript := func(dest []*messages.Resource, script *Script) []*messages.Resource {
			res := new(messages.Resource)
			res.Name = proto.String(script.Path)
			kind := messages.ResourceType_RType_Script
			res.Type = &kind
			res.Script = proto.String(string(script.Source))
			res.ShouldExecSCript = proto.Bool(script.ShouldExec)
			dest = append(dest, res)
			return dest
		}

		// get all resources
		resources := make([]*messages.Resource, 0)
		for _, addon := range e.Addons {
			for _, script := range addon.ClientScripts {
				resources = appendResourceScript(resources, script)
			}
		}
		response := new(messages.GetCLResourcesResp)
		response.Resources = resources
		encoded, err := netengine.EncodeMessage(response, int32(messages.MessageTypes_GETCLRESOURCESRESP))
		if err != nil {
			e.Log.Error("Error encoding GetClResources response ", err)
			return make([]byte, 0)

		}
		return encoded
	}
}

//处理查找节点的请求
//本节点定期查询已知节点是否在线，更新节点信息
func (this *Manager) read() {
	clientConn, _ := this.serverManager.GetController().GetSession(this.rootId.String())
	for {
		node := <-this.nodeStore.OutFindNode
		if node.NodeId != nil {
			findNodeOne := &msg.FindNodeReq{
				NodeId: proto.String(this.nodeStore.GetRootId()),
				FindId: proto.String(node.NodeId.String()),
			}
			findNodeBytes, _ := proto.Marshal(findNodeOne)
			// clientConn := this.serverManager.GetController().GetClientByName("firstConnPeer")
			// fmt.Println(clientConn)
			clientConn.Send(msg.FindNodeReqNum, &findNodeBytes)
		}
		if node.NodeIdShould != nil {
			findNodeOne := &msg.FindNodeReq{
				NodeId: proto.String(this.nodeStore.GetRootId()),
				FindId: proto.String(node.NodeIdShould.String()),
			}
			findNodeBytes, _ := proto.Marshal(findNodeOne)
			// clientConn := this.serverManager.GetController().GetClientByName("firstConnPeer")
			clientConn.Send(msg.FindNodeReqNum, &findNodeBytes)
		}
	}
}

func TestResourceOffersMessageHandling(t *testing.T) {
	sched := NewMesosScheduler()
	sched.ResourceOffers = func(driver *SchedulerDriver, offers []*mesos.Offer) {
		if len(offers) != 1 {
			log.Fatalf("Scheduler.ResourceOffers expected 1 offer, but got", len(offers))
		}
		if offers[0].GetId().GetValue() != "offer-1" {
			log.Fatalln("Scheduler.ResourceOffers exepected value not received")
		}
	}

	msg := &mesos.ResourceOffersMessage{
		Offers: []*mesos.Offer{
			&mesos.Offer{
				Id:          &mesos.OfferID{Value: proto.String("offer-1")},
				FrameworkId: &mesos.FrameworkID{Value: proto.String("test-framework-1")},
				SlaveId:     &mesos.SlaveID{Value: proto.String("test-slave-1")},
				Hostname:    proto.String("localhost"),
			},
		},
	}
	driver, err := NewSchedDriver(sched, &mesos.FrameworkInfo{}, "localhost:0")
	if err != nil {
		t.Fatal(err)
	}
	driver.schedMsgQ <- msg
}

func TestParseSectionAddressTable_Valid(t *testing.T) {
	lines := []string{
		"10 11",
		"MAC1 IP1",
		"MAC2 IP2",
	}
	trace := Trace{}
	err := parseSectionAddressTable(lines, &trace)
	if err != nil {
		t.Fatal("Unexpected error:", err)
	}
	expectedTrace := Trace{
		AddressTableFirstId: proto.Int32(10),
		AddressTableSize:    proto.Int32(11),
		AddressTableEntry: []*AddressTableEntry{
			&AddressTableEntry{
				MacAddress: proto.String("MAC1"),
				IpAddress:  proto.String("IP1"),
			},
			&AddressTableEntry{
				MacAddress: proto.String("MAC2"),
				IpAddress:  proto.String("IP2"),
			},
		},
	}
	checkProtosEqual(t, &expectedTrace, &trace)
}

func (p *Parser) Parse() *tp.ScriptObject {
	script := new(tp.ScriptObject)
	// script.Name = proto.String(p.FullPath)

	if !p.RootFile || TritiumParserShowRewriterFileName {
		script.Name = proto.String(filepath.Join(p.ScriptPath, p.FileName))
	} else {
		script.Name = proto.String("__rewriter__")
	}

	stmts := tp.ListInstructions()
	defs := make([]*tp.Function, 0) // Add a new constructor in instruction.go

	// Look for the namespace directive first.
	if p.peek().Lexeme == NAMESPACE {
		p.namespaces()
	}

	for p.peek().Lexeme != EOF {
		switch p.peek().Lexeme {
		case FUNC:
			defs = append(defs, p.definition())
		case OPEN:
			p.open(false)
		default:
			stmt := p.statement()
			stmts = append(stmts, stmt)
			// need to intersperse imports with definitions
			if constants.Instruction_InstructionType_name[int32(stmt.GetType())] == "IMPORT" {
				// Make a special function stub that represents the import.
				// Need to do this because we can't mix definitions and instructions in
				// the same array.
				imp := new(tp.Function)
				imp.Name = proto.String("@import")
				imp.Description = proto.String(stmt.GetValue())
				defs = append(defs, imp)
			}
		}
	}

	if len(defs) == 0 {
		defs = nil
	}

	var line int32
	if len(stmts) == 0 {
		stmts = nil
	} else {
		line = *stmts[0].LineNumber
	}

	script.Functions = defs
	script.Root = tp.MakeBlock(stmts, line)

	// if defs == nil && p.currentNamespace() != "tritium" {
	// 	panic(fmt.Sprintf("%s: %d -- custom modules may only be declared in function definition files", p.FileName, moduleLineNum))
	// }

	return script
}

func (g *Gossiper) gossipTo(addr *net.TCPAddr, digests []*Digest) {
	conn, err := net.DialTCP("tcp", nil, addr)
	if err != nil {
		g.logger.Printf("gossiper gossip to %s error: %s\n", addr, err)
		return
	}

	g.mutex.Lock()
	if g.stoped {
		g.mutex.Unlock()
		conn.Close()
		return
	}

	syn := &SynMessage{
		Id:      proto.String(g.id),
		Origin:  proto.String(g.origin),
		Digests: digests,
	}

	g.mutex.Unlock()

	if err := encode(conn, syn); err != nil {
		conn.Close()
	}

	g.readMessage(conn)
}

func (g *Gossiper) handleSynMessage(conn net.Conn, syn *SynMessage) {
	if g.origin != syn.GetOrigin() {
		conn.Close()
		return
	}

	g.mutex.Lock()
	if g.stoped {
		g.mutex.Unlock()
		conn.Close()
		return
	}

	digests, deltas := g.scuttlebutt(syn.GetId(), syn.GetDigests())

	ack := &AckMessage{
		Id:      proto.String(g.id),
		Origin:  proto.String(g.origin),
		Digests: digests,
		Deltas:  deltas,
	}

	g.mutex.Unlock()

	if err := encode(conn, ack); err != nil {
		conn.Close()
	}

	g.readMessage(conn)
}

// keyToProto converts a *Key to a Reference proto.
func keyToProto(defaultAppID string, k *Key) *pb.Reference {
	appID := k.appID
	if appID == "" {
		appID = defaultAppID
	}
	n := 0
	for i := k; i != nil; i = i.parent {
		n++
	}
	e := make([]*pb.Path_Element, n)
	for i := k; i != nil; i = i.parent {
		n--
		e[n] = &pb.Path_Element{
			Type: &i.kind,
		}
		// At most one of {Name,Id} should be set.
		// Neither will be set for incomplete keys.
		if i.stringID != "" {
			e[n].Name = &i.stringID
		} else if i.intID != 0 {
			e[n].Id = &i.intID
		}
	}
	var namespace *string
	if k.namespace != "" {
		namespace = proto.String(k.namespace)
	}
	return &pb.Reference{
		App:       proto.String(appID),
		NameSpace: namespace,
		Path: &pb.Path{
			Element: e,
		},
	}
}

// Encodes the AppendEntriesRequest to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (req *AppendEntriesRequest) Encode(w io.Writer) (int, error) {

	protoEntries := make([]*protobuf.ProtoAppendEntriesRequest_ProtoLogEntry, len(req.Entries))

	for i, entry := range req.Entries {
		protoEntries[i] = &protobuf.ProtoAppendEntriesRequest_ProtoLogEntry{
			Index:       proto.Uint64(entry.Index),
			Term:        proto.Uint64(entry.Term),
			CommandName: proto.String(entry.CommandName),
			Command:     entry.Command,
		}
	}

	pb := &protobuf.ProtoAppendEntriesRequest{
		Term:         proto.Uint64(req.Term),
		PrevLogIndex: proto.Uint64(req.PrevLogIndex),
		PrevLogTerm:  proto.Uint64(req.PrevLogTerm),
		CommitIndex:  proto.Uint64(req.CommitIndex),
		LeaderName:   proto.String(req.LeaderName),
		Entries:      protoEntries,
	}

	p, err := proto.Marshal(pb)
	if err != nil {
		return -1, err
	}

	return w.Write(p)
}

func (self *ResMan) Run(master string) {
	frameworkIdStr := FRAMEWORK_ID
	frameworkId := &mesos.FrameworkID{Value: &frameworkIdStr}
	driver := mesos.SchedulerDriver{
		Master: master,
		Framework: mesos.FrameworkInfo{
			Name:            proto.String("TyrantFramework"),
			User:            proto.String(""),
			FailoverTimeout: failoverTimeout,
			Id:              frameworkId,
		},

		Scheduler: &mesos.Scheduler{
			ResourceOffers: self.OnResourceOffers,
			StatusUpdate:   self.OnStatusUpdate,
			Error:          self.OnError,
			Disconnected:   self.OnDisconnected,
			Registered:     self.OnRegister,
			Reregistered:   self.OnReregister,
		},
	}

	driver.Init()
	defer driver.Destroy()
	go self.EventLoop()

	driver.Start()
	<-self.exit
	log.Debug("exit")
	driver.Stop(false)
}

func testTargetManager(t test.Tester) {
	targetManager := NewTargetManager(nopIngester{}, 3)
	testJob1 := config.JobConfig{
		JobConfig: pb.JobConfig{
			Name:           proto.String("test_job1"),
			ScrapeInterval: proto.String("1m"),
		},
	}
	testJob2 := config.JobConfig{
		JobConfig: pb.JobConfig{
			Name:           proto.String("test_job2"),
			ScrapeInterval: proto.String("1m"),
		},
	}

	target1GroupA := &fakeTarget{
		schedules: []time.Time{time.Now()},
		interval:  time.Minute,
	}
	target2GroupA := &fakeTarget{
		schedules: []time.Time{time.Now()},
		interval:  time.Minute,
	}

	targetManager.AddTarget(testJob1, target1GroupA)
	targetManager.AddTarget(testJob1, target2GroupA)

	target1GroupB := &fakeTarget{
		schedules: []time.Time{time.Now()},
		interval:  time.Minute * 2,
	}

	targetManager.AddTarget(testJob2, target1GroupB)
}

func main() {
	log.Println("Starting gomestest.")
	log.Println("Assuming master 127.0.0.1:5050...")

	master := "127.0.0.1:5050"

	framework := &mesos.FrameworkInfo{
		User: proto.String("test"),
		Name: proto.String("gomes"),
		Id:   &mesos.FrameworkID{Value: proto.String("gomes-framework-1")},
	}

	log.Println("Registering framework" + framework.String())

	driver, err := gomes.NewSchedDriver(Sched, framework, master)
	if err != nil {
		log.Println("Unable to create a SchedulerDriver", err.Error())
	}

	stat := driver.Run()
	if stat != mesos.Status_DRIVER_STOPPED {
		log.Println("A problem occured, framework reported status " + stat)
	}
	driver.Stop(false)
}

func TestParseSectionIntro_Valid(t *testing.T) {
	lines := []string{
		"10",
		"BUILDID",
		"NODEID 123 321 789",
		"12 23 34",
	}
	trace := Trace{}
	err := parseSectionIntro(lines, &trace)
	if err != nil {
		t.Fatal("Unexpected error:", err)
	}
	expectedTrace := Trace{
		FileFormatVersion: proto.Int32(10),
		BuildId:           proto.String("BUILDID"),
		NodeId:            proto.String("NODEID"),
		ProcessStartTimeMicroseconds: proto.Int64(123),
		SequenceNumber:               proto.Int32(321),
		TraceCreationTimestamp:       proto.Int64(789),
		PcapReceived:                 proto.Uint32(12),
		PcapDropped:                  proto.Uint32(23),
		InterfaceDropped:             proto.Uint32(34),
	}
	checkProtosEqual(t, &expectedTrace, &trace)
}

func (g *Gossiper) handleAckMessage(conn net.Conn, ack *AckMessage) {
	if g.origin != ack.GetOrigin() {
		conn.Close()
		return
	}

	g.mutex.Lock()
	if g.stoped {
		g.mutex.Unlock()
		conn.Close()
		return
	}

	g.updateDeltas(ack.GetDeltas())

	ack2 := &Ack2Message{
		Id:     proto.String(g.id),
		Origin: proto.String(g.origin),
		Deltas: g.fetchDeltas(ack.GetDigests()),
	}

	g.mutex.Unlock()

	if err := encode(conn, ack2); err != nil {
		conn.Close()
	}

	g.readMessage(conn)
}

func TestParseSectionDnsTableA_Valid(t *testing.T) {
	lines := []string{
		"10 11",
		"12 13 0 DOM1 IP1 14",
		"15 16 1 DOM2 IP2 17",
	}
	trace := Trace{}
	err := parseSectionDnsTableA(lines, &trace)
	if err != nil {
		t.Fatal("Unexpected error:", err)
	}
	expectedTrace := Trace{
		ARecordsDropped:     proto.Int32(10),
		CnameRecordsDropped: proto.Int32(11),
		ARecord: []*DnsARecord{
			&DnsARecord{
				PacketId:   proto.Int32(12),
				AddressId:  proto.Int32(13),
				Anonymized: proto.Bool(false),
				Domain:     proto.String("DOM1"),
				IpAddress:  proto.String("IP1"),
				Ttl:        proto.Int32(14),
			},
			&DnsARecord{
				PacketId:   proto.Int32(15),
				AddressId:  proto.Int32(16),
				Anonymized: proto.Bool(true),
				Domain:     proto.String("DOM2"),
				IpAddress:  proto.String("IP2"),
				Ttl:        proto.Int32(17),
			},
		},
	}
	checkProtosEqual(t, &expectedTrace, &trace)
}

func MsgProjectDelete(conn net.Conn, msg *message.Msg) {
	proj := &Project{
		Id: *msg.Projects.Id,
	}
	var answer *message.Msg
	temp := getListOfUser(proj.Id)
	if err := proj.Del(dbPool); err != nil {
		answer = &message.Msg{
			Target:    message.TARGET_PROJECTS.Enum(),
			Command:   message.CMD_ERROR.Enum(),
			AuthorId:  proto.Uint32(*msg.AuthorId),
			SessionId: proto.String(*msg.SessionId),
			Error: &message.Msg_Error{
				ErrorId: proto.Uint32(32),
			},
		}
	} else {
		answer = &message.Msg{
			Target:    message.TARGET_PROJECTS.Enum(),
			Command:   message.CMD_SUCCES.Enum(),
			AuthorId:  proto.Uint32(*msg.AuthorId),
			SessionId: proto.String(*msg.SessionId),
		}
		sendMsgToallUser(temp, msg)
	}
	data, err := proto.Marshal(answer)
	if err != nil {
		LOGGER.Print("Impossible to marshal msg in MsgProjectUpdate", err, answer)
		return
	}
	conn.Write(write_int32(int32(len(data))))
	conn.Write(data)
}

func main() {
	test := &example.Test{
		Label: proto.String("hello"),
		Type:  proto.Int32(17),
		Optionalgroup: &example.Test_OptionalGroup{
			RequiredField: proto.String("good bye"),
		},
	}
	log.Println(test)
	data, err := proto.Marshal(test)
	log.Println(data, err)
	if err != nil {
		log.Fatal("marshaling error: ", err)
	}
	newTest := &example.Test{}
	err = proto.Unmarshal(data, newTest)
	if err != nil {
		log.Fatal("unmarshaling error: ", err)
	}
	// Now test and newTest contain the same data.
	log.Println(test.GetLabel(), newTest.GetLabel())
	if test.GetLabel() != newTest.GetLabel() {
		log.Fatalf("data mismatch %q != %q", test.GetLabel(), newTest.GetLabel())
	}
	// etc.
}

func SampleToMetricDTO(s *Sample) *dto.Metric {
	labelLength := len(s.Metric)
	labelNames := make([]string, 0, labelLength)

	for labelName := range s.Metric {
		labelNames = append(labelNames, string(labelName))
	}

	sort.Strings(labelNames)

	labelSets := make([]*dto.LabelPair, 0, labelLength)

	for _, labelName := range labelNames {
		labelValue := s.Metric[LabelName(labelName)]
		labelPair := &dto.LabelPair{
			Name:  proto.String(string(labelName)),
			Value: proto.String(string(labelValue)),
		}

		labelSets = append(labelSets, labelPair)
	}

	return &dto.Metric{
		LabelPair: labelSets,
	}
}