func (pc *PacketInjectorClient) injectPacket(host string, pp *PacketParams, status chan *string) {
	msg := shttp.NewWSMessage(Namespace, "InjectPacket", pp)

	ch := make(chan *json.RawMessage)
	defer close(ch)

	pc.replyChanMutex.Lock()
	pc.replyChan[msg.UUID] = ch
	pc.replyChanMutex.Unlock()

	defer func() {
		pc.replyChanMutex.Lock()
		delete(pc.replyChan, msg.UUID)
		pc.replyChanMutex.Unlock()
	}()

	if !pc.WSServer.SendWSMessageTo(msg, host) {
		e := fmt.Sprintf("Unable to send message to agent: %s", host)
		logging.GetLogger().Errorf(e)
		status <- &e
		return
	}

	data := <-ch
	var reply string
	if err := json.Unmarshal([]byte(*data), &reply); err != nil {
		e := fmt.Sprintf("Error while reading reply from: %s", host)
		logging.GetLogger().Errorf(e)
		status <- &e
		return
	}

	status <- &reply
}

func (f *TableClient) lookupFlows(flowset chan *FlowSet, host string, flowSearchQuery *FlowSearchQuery) {
	obj, _ := proto.Marshal(flowSearchQuery)
	tq := TableQuery{
		Type: "FlowSearchQuery",
		Obj:  obj,
	}
	msg := shttp.NewWSMessage(Namespace, "TableQuery", tq)

	ch := make(chan *json.RawMessage)
	defer close(ch)

	f.replyChanMutex.Lock()
	f.replyChan[msg.UUID] = ch
	f.replyChanMutex.Unlock()

	defer func() {
		f.replyChanMutex.Lock()
		delete(f.replyChan, msg.UUID)
		f.replyChanMutex.Unlock()
	}()

	if !f.WSServer.SendWSMessageTo(msg, host) {
		logging.GetLogger().Errorf("Unable to send message to agent: %s", host)
		flowset <- NewFlowSet()
		return
	}

	select {
	case raw := <-ch:
		var reply TableReply
		if err := json.Unmarshal([]byte(*raw), &reply); err != nil {
			logging.GetLogger().Errorf("Error returned while reading TableReply from: %s", host)
			break
		}

		if reply.Status != http.StatusOK {
			logging.GetLogger().Errorf("Error %d TableReply from: %s", reply.Status, host)
			break
		}

		fs := NewFlowSet()

		context := MergeContext{Sorted: flowSearchQuery.Sort, Dedup: flowSearchQuery.Dedup}
		for _, b := range reply.Obj {
			var fsr FlowSearchReply
			if err := proto.Unmarshal(b, &fsr); err != nil {
				logging.GetLogger().Errorf("Unable to decode flow search reply from: %s", host)
				continue
			}
			fs.Merge(fsr.FlowSet, context)
		}
		flowset <- fs

		return
	case <-time.After(time.Second * 10):
		logging.GetLogger().Errorf("Timeout while reading TableReply from: %s", host)
	}

	flowset <- NewFlowSet()
}

func (o *OnDemandProbeClient) unregisterProbe(node *graph.Node) bool {
	msg := shttp.NewWSMessage(ondemand.Namespace, "CaptureStop", ondemand.CaptureQuery{NodeID: string(node.ID)})

	if !o.wsServer.SendWSMessageTo(msg, node.Host()) {
		logging.GetLogger().Errorf("Unable to send message to agent: %s", node.Host())
		return false
	}

	return true
}

func (s *GraphServer) OnMessage(c *shttp.WSClient, msg shttp.WSMessage) {
	if msg.Namespace != Namespace {
		return
	}

	s.Graph.Lock()
	defer s.Graph.Unlock()

	msgType, obj, err := UnmarshalWSMessage(msg)
	if err != nil {
		logging.GetLogger().Errorf("Graph: Unable to parse the event %v: %s", msg, err.Error())
		return
	}

	switch msgType {
	case "SyncRequest":
		reply := shttp.NewWSMessage(Namespace, "SyncReply", s.Graph.WithContext(obj.(GraphContext)))
		c.SendWSMessage(reply)
	case "SubGraphDeleted":
		n := obj.(*Node)

		logging.GetLogger().Debugf("Got SubGraphDeleted event from the node %s", n.ID)

		node := s.Graph.GetNode(n.ID)
		if node != nil {
			s.Graph.DelSubGraph(node)
		}
	case "NodeUpdated":
		n := obj.(*Node)
		node := s.Graph.GetNode(n.ID)
		if node != nil {
			s.Graph.SetMetadata(node, n.metadata)
		}
	case "NodeDeleted":
		s.Graph.DelNode(obj.(*Node))
	case "NodeAdded":
		n := obj.(*Node)
		if s.Graph.GetNode(n.ID) == nil {
			s.Graph.AddNode(n)
		}
	case "EdgeUpdated":
		e := obj.(*Edge)
		edge := s.Graph.GetEdge(e.ID)
		if edge != nil {
			s.Graph.SetMetadata(edge, e.metadata)
		}
	case "EdgeDeleted":
		s.Graph.DelEdge(obj.(*Edge))
	case "EdgeAdded":
		e := obj.(*Edge)
		if s.Graph.GetEdge(e.ID) == nil {
			s.Graph.AddEdge(e)
		}
	}
}

func (c *Forwarder) triggerResync() {
	logging.GetLogger().Infof("Start a resync of the graph")

	c.Graph.RLock()
	defer c.Graph.RUnlock()

	// request for deletion of everything belonging to Root node
	c.Client.SendWSMessage(shttp.NewWSMessage(Namespace, "HostGraphDeleted", c.Host))

	// re-added all the nodes and edges
	nodes := c.Graph.GetNodes(Metadata{})
	for _, n := range nodes {
		c.Client.SendWSMessage(shttp.NewWSMessage(Namespace, "NodeAdded", n))
	}

	edges := c.Graph.GetEdges(Metadata{})
	for _, e := range edges {
		c.Client.SendWSMessage(shttp.NewWSMessage(Namespace, "EdgeAdded", e))
	}
}

func (mapper *NeutronMapper) updateNode(node *graph.Node, attrs *Attributes) {
	mapper.graph.Lock()

	tr := mapper.graph.StartMetadataTransaction(node)
	tr.AddMetadata("Manager", "neutron")

	if attrs.PortID != "" {
		tr.AddMetadata("Neutron/PortID", attrs.PortID)
	}

	if attrs.TenantID != "" {
		tr.AddMetadata("Neutron/TenantID", attrs.TenantID)
	}

	if attrs.NetworkID != "" {
		tr.AddMetadata("Neutron/NetworkID", attrs.NetworkID)
	}

	if attrs.NetworkName != "" {
		tr.AddMetadata("Neutron/NetworkName", attrs.NetworkName)
	}

	if segID, err := strconv.Atoi(attrs.VNI); err != nil && segID > 0 {
		tr.AddMetadata("Neutron/VNI", uint64(segID))
	}

	var registerLink *FabricRegisterLinkWSMessage
	if vm, ok := node.Metadata()["ExtID/vm-uuid"]; ok {
		if mac, ok := node.Metadata()["ExtID/attached-mac"]; ok {
			if path := mapper.graph.LookupShortestPath(node, graph.Metadata{"Type": "tun"}, graph.Metadata{"RelationType": "layer2"}); len(path) > 0 {
				registerLink = &FabricRegisterLinkWSMessage{
					ParentNodeID:   path[len(path)-1].ID,
					ParentMetadata: graph.Metadata{"Type": "host", "InstanceID": vm},
					ChildMetadata:  graph.Metadata{"Type": "device", "MAC": mac},
				}
			}
		}
	}
	tr.Commit()
	mapper.graph.Unlock()

	if registerLink != nil {
		msg := shttp.NewWSMessage(FabricNamespace, "RegisterLink", registerLink)
		mapper.wsClient.SendWSMessage(msg)
	}
}

/* Called by alert.EvalNodes() */
func (c *alertClient) OnAlert(amsg *AlertMessage) {
	msg := shttp.NewWSMessage(Namespace, "Alert", amsg)
	c.wsClient.SendWSMessage(msg)
}

func (c *Forwarder) OnEdgeDeleted(e *Edge) {
	c.Client.SendWSMessage(shttp.NewWSMessage(Namespace, "EdgeDeleted", e))
}

func (c *Forwarder) OnNodeDeleted(n *Node) {
	c.Client.SendWSMessage(shttp.NewWSMessage(Namespace, "NodeDeleted", n))
}

func (s *GraphServer) OnEdgeDeleted(e *Edge) {
	s.WSServer.BroadcastWSMessage(shttp.NewWSMessage(Namespace, "EdgeDeleted", e))
}

func (s *GraphServer) OnNodeDeleted(n *Node) {
	s.WSServer.BroadcastWSMessage(shttp.NewWSMessage(Namespace, "NodeDeleted", n))
}