package message
import (
"container/list"
"data_manager"
"encoding/json"
"errors"
"fmt"
"sync"
"github.com/pojoin/golis"
)
var once sync.Once
type CommandProperty int
type Command interface {
//BuildResponse(info interface{}) (string, error)
CommandHandle(session *golis.Iosession, CmdInfo interface{}) error
GetCommandName() string
IsProvidedResult() bool
}
type CommandAttribute struct {
CommandName string
IsProvidedResult bool //CommandProperty
}
////////////////////////////dmc command ////////////////////////////
////////////////////// dmc_register /////////////////////
type DmcRegist struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
RoomId int32 `json:"room_id"`
}
type DmcCommandRegist struct {
cmdAtt CommandAttribute
}
////////////////////// dmc_heart_beat //////////////////
type DmcHeartBeat struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
DmcId int32 `json:"dmc_id"`
}
type DmcCommandHeartBeat struct {
cmdAtt CommandAttribute
}
////////////////////// dmc_unregister //////////////////
type DmcUnregister struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
DmcId int32 `json:"dmc_id"`
}
type DmcCommandUnregister struct {
cmdAtt CommandAttribute
}
////////////////////// dmc_stream_query //////////////////
type DmcStreamQuery struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
DmcId int32 `json:"dmc_id"`
StreamName string `json:"stream_name"`
}
type DmcCommandStreamQuery struct {
cmdAtt CommandAttribute
}
////////////////// dmc _all_stream_update ////////////////
type DmcAllStreamUpdate struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
DmcId int32 `json:"dmc_id"`
StreamInfoList []data_manager.StreamAttributes `json:"stream_info"`
}
type DmcCommandAllStreamUpdate struct {
cmdAtt CommandAttribute
}
////////////////// dmc _capability_update ////////////////
type DmcCapabilityUpdate struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
DmcId int32 `json:"dmc_id"`
Cap Capability `json:"capability"`
NCap NetworkCapability `json:"network_capability"`
Ld Load `json:"load"`
NLd NetworkLoad `json:"network_load"`
}
type DmcCommandCapabilityUpdate struct {
cmdAtt CommandAttribute
}
//////////////////// dmc _stream_ update ////////////////
type DmcStreamUpdate struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
DmcId int32 `json:"dmc_id"`
////////////////// StreamAttributes /////////////////
Operate data_manager.StreamOperate `json:"operate"`
Name string `json:"stream_name"`
Status data_manager.StreamState `json:"status"`
Attributes data_manager.StreamAttribute `json:"attributes"`
Type data_manager.StreamType `json:"type"`
DmsInfo []string `json:"dms_info"`
}
type DmcCommandStreamUpdate struct {
cmdAtt CommandAttribute
}
//////////////////// dmc_get_dms_request ////////////////
type DmcGetDmsRequest struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
DmcId int32 `json:"dmc_id"`
Type data_manager.StreamType `json:"type"`
}
type DmcCommandGetDmsRequest struct {
cmdAtt CommandAttribute
}
////////// dmc_get_dms_response /////////////
type DmcGetDmsResponse struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
DmcId int32 `json:"dmc_id"`
DmsInfo []string `json:"dms_info"`
}
type DmcCommandGetDmsResponse struct {
cmdAtt CommandAttribute
}
//////////////////////////////////////////////////////////////////
////////////////////////// gmc request & response /////////////////////////////
func AddFrameTail(input []byte) []byte {
buf := make([]byte, len(input)+1)
copy(buf, input[:])
buf[len(input)] = 0x00
return buf
}
func JsonMarshal(msgStruct interface{}) ([]byte, error) {
b, err := json.Marshal(msgStruct)
if err != nil {
fmt.Println("encoding faild")
return nil, errors.New("encoding faild")
} else {
fmt.Println("encoded data : ")
//fmt.Println(b)
fmt.Println(string(b))
}
return b, err
}
type GmcMessage interface {
BuildMessage() ([]byte, error)
}
/////// gmc_reg_response ///////
type GmcRegResponse struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
DmcId int32 `json:"dmc_id"`
}
func (gm *GmcRegResponse) BuildMessage() ([]byte, error) {
//gmcRegResponse := &GmcRegResponse{"gmc_reg_response", gm.MsgId, gm.DmcId}
b, err := JsonMarshal(*gm)
if b == nil {
return b, err
}
buf := AddFrameTail(b)
return buf, nil
}
/////// gmc_response ///////
type GmcResponse struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
}
func (gm *GmcResponse) BuildMessage() ([]byte, error) {
//gmcResponse := &GmcResponse{"gmc_response", gm.MsgId}
b, err := JsonMarshal(*gm)
if b == nil {
return b, err
}
buf := AddFrameTail(b)
return buf, nil
}
/////// gmc_stream_query_result ///////
type GmcStreamQueryResult struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
Result string `json:"result"`
StreamName string `json:"stream_name"`
DmsInfo *list.List `json:"dms_info"`
}
func (gm *GmcStreamQueryResult) BuildMessage() ([]byte, error) {
b, err := JsonMarshal(*gm)
if b == nil {
return b, err
}
buf := AddFrameTail(b)
return buf, nil
}
//////// gmc_get_dms_request //////////
type GmcGetDmsRequest struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
}
func (gm *GmcGetDmsRequest) BuildMessage() ([]byte, error) {
//gmcResponse := &GmcGetDmsRequest{"gmc_get_dms_request", gm.MsgId}
b, err := JsonMarshal(*gm)
if b == nil {
return b, err
}
buf := AddFrameTail(b)
return buf, nil
}
/////////// gmc_get_dms_response /////////
type GmcGetDmsResponse struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
DmcId int32 `json:"dmc_id"`
DmsInfo []string `json:"dms_info"`
}
func (gm *GmcGetDmsResponse) BuildMessage() ([]byte, error) {
b, err := JsonMarshal(*gm)
if b == nil {
return b, err
}
buf := AddFrameTail(b)
return buf, nil
}
/////// gmc_stream_published_notify //////
type GmcStreamPublishedNotify struct {
Command string `json:"command"`
MsgId string `json:"msg_id"`
StreamName string `json:"stream_name"`
DmsInfo []string `json:"dms_info"`
}
func (gm *GmcStreamPublishedNotify) BuildMessage() ([]byte, error) {
b, err := JsonMarshal(*gm)
if b == nil {
return b, err
}
buf := AddFrameTail(b)
return buf, nil
}
/////////// func BuildGmcMessage ////////////
func BuildGmcMessage(gmcMsg GmcMessage) ([]byte, error) {
msg, err := gmcMsg.BuildMessage()
if err != nil {
return nil, err
}
return msg, nil
}
///////////////////////////////////////////////////////////////////////////////////
func (dcr *DmcCommandRegist) CommandHandle(session *golis.Iosession, CmdInfo interface{}) error {
fmt.Println("register command handle ........ ")
cmd := CmdInfo.(*DmcRegist)
fmt.Println("cmd:", cmd.Command, " msgid:", cmd.MsgId, " roomid:", cmd.RoomId)
dmc_instance := data_manager.GetDmcInstance()
dmc_id := dmc_instance.GetDmcId()
dmc_info := data_manager.DmcInfo{dmc_id, list.New(), list.New(), cmd.RoomId, session}
dmc_instance.AddDmcInfo(dmc_id, dmc_info)
gmcResponse := &GmcRegResponse{"gmc_reg_response", cmd.MsgId, dmc_id}
buf, err := BuildGmcMessage(gmcResponse)
if err != nil {
return err
}
fmt.Println("------------- Session write frame: ---")
session.Write(buf)
return nil
}
func (dcr *DmcCommandRegist) GetCommandName() string {
return dcr.cmdAtt.CommandName
}
func (dcr *DmcCommandRegist) IsProvidedResult() bool {
return dcr.cmdAtt.IsProvidedResult
}
///////////////////// dmc_heart_beat ////////////////////////
func (dcr *DmcCommandHeartBeat) CommandHandle(session *golis.Iosession, CmdInfo interface{}) error {
fmt.Println("heart beat command handle ........ ")
cmd := CmdInfo.(*DmcHeartBeat)
fmt.Println("cmd:", cmd.Command, " msgid:", cmd.MsgId, " dmcid:", cmd.DmcId)
return nil
}
func (dcr *DmcCommandHeartBeat) GetCommandName() string {
return dcr.cmdAtt.CommandName
}
func (dcr *DmcCommandHeartBeat) IsProvidedResult() bool {
return dcr.cmdAtt.IsProvidedResult
}
////////////////////// dmc_unregister //////////////////
func (dcr *DmcCommandUnregister) CommandHandle(session *golis.Iosession, CmdInfo interface{}) error {
fmt.Println("unregister command handle ........ ")
cmd := CmdInfo.(*DmcUnregister)
fmt.Println("cmd:", cmd.Command, " msgid:", cmd.MsgId, " dmcid:", cmd.DmcId)
dmc_instance := data_manager.GetDmcInstance()
dmc_instance.DelDmcInfo(cmd.DmcId)
gmcResponse := &GmcResponse{"gmc_response", cmd.MsgId}
buf, err := BuildGmcMessage(gmcResponse)
if err != nil {
return err
}
fmt.Println("------------- Session write frame: ---")
session.Write(buf)
return nil
}
func (dcr *DmcCommandUnregister) GetCommandName() string {
return dcr.cmdAtt.CommandName
}
func (dcr *DmcCommandUnregister) IsProvidedResult() bool {
return dcr.cmdAtt.IsProvidedResult
}
////////////////////// dmc_stream_query //////////////////
func (dcr *DmcCommandStreamQuery) CommandHandle(session *golis.Iosession, CmdInfo interface{}) error {
fmt.Println("stream query command handle ........ ")
cmd := CmdInfo.(*DmcStreamQuery)
fmt.Println("cmd:", cmd.Command, " msgid:", cmd.MsgId, " dmcid:", cmd.DmcId, "streamname:", cmd.StreamName)
stream_instance := data_manager.GetStreamInstance()
stream_info, result := stream_instance.GetStreamInfo(cmd.StreamName)
var queryresponse *GmcStreamQueryResult
if result == true {
queryresponse = &GmcStreamQueryResult{"gmc_stream_query_result", cmd.MsgId, "Found", cmd.StreamName, stream_info.DmsInfo}
} else {
queryresponse = &GmcStreamQueryResult{"gmc_stream_query_result", cmd.MsgId, "Not Found", cmd.StreamName, nil}
}
b, err := json.Marshal(queryresponse)
if err != nil {
fmt.Println("encoding faild")
return errors.New("encoding faild")
} else {
fmt.Println("encoded data : ")
//fmt.Println(b)
fmt.Println(string(b))
}
buf := make([]byte, len(b)+1)
copy(buf, b[:])
buf[len(b)] = 0x00
fmt.Println("------------- Session write frame: ---")
session.Write(buf)
return nil
}
func (dcr *DmcCommandStreamQuery) GetCommandName() string {
return dcr.cmdAtt.CommandName
}
func (dcr *DmcCommandStreamQuery) IsProvidedResult() bool {
return dcr.cmdAtt.IsProvidedResult
}
////////////////// dmc_all_stream_update ////////////////
func (dcr *DmcCommandAllStreamUpdate) CommandHandle(session *golis.Iosession, CmdInfo interface{}) error {
fmt.Println("all stream update command handle ........ ")
cmd := CmdInfo.(*DmcAllStreamUpdate)
fmt.Printf("------------------ len=%d cap=%d slice=%v\n", len(cmd.StreamInfoList), cap(cmd.StreamInfoList), cmd.StreamInfoList)
fmt.Println("cmd:", cmd.Command, " dmc id:", cmd.DmcId, " msg id:", cmd.MsgId)
stream_instance := data_manager.GetStreamInstance()
stream_lst := cmd.StreamInfoList
for _, stream_info := range stream_lst {
temp_stream_info := data_manager.NewStreamInfo(stream_info.Name, cmd.DmcId, stream_info.State, stream_info.Attributes, stream_info.Type, list.New())
stream_instance.AddStreamInfo(stream_info.Name, temp_stream_info)
}
gmcResponse := &GmcResponse{"gmc_response", cmd.MsgId}
buf, err := BuildGmcMessage(gmcResponse)
if err != nil {
return err
}
fmt.Println("------------- Session write frame: ---")
session.Write(buf)
return nil
}
func (dcr *DmcCommandAllStreamUpdate) GetCommandName() string {
return dcr.cmdAtt.CommandName
}
func (dcr *DmcCommandAllStreamUpdate) IsProvidedResult() bool {
return dcr.cmdAtt.IsProvidedResult
}
////////////////// dmc_capability_update ////////////////
func (dcr *DmcCommandCapabilityUpdate) CommandHandle(session *golis.Iosession, CmdInfo interface{}) error {
fmt.Println("capability update command handle ........ ")
cmd := CmdInfo.(*DmcCapabilityUpdate)
fmt.Println("cmd:", cmd.Command, " msgid:", cmd.MsgId, " dmcid:", cmd.DmcId, " cap item:", cmd.Cap.Item, " ld item", cmd.Ld.Item, " nld item:", cmd.NLd.Item)
return nil
}
func (dcr *DmcCommandCapabilityUpdate) GetCommandName() string {
return dcr.cmdAtt.CommandName
}
func (dcr *DmcCommandCapabilityUpdate) IsProvidedResult() bool {
return dcr.cmdAtt.IsProvidedResult
}
//////////////////// dmc_stream_update ////////////////
func (dcr *DmcCommandStreamUpdate) CommandHandle(session *golis.Iosession, CmdInfo interface{}) error {
fmt.Println("stream update command handle ........ ")
cmd := CmdInfo.(*DmcStreamUpdate)
fmt.Println(" cmd:", cmd.Command, " dmc id:", cmd.DmcId, " msg id:", cmd.MsgId, " stream name:", cmd.Name)
stream_instance := data_manager.GetStreamInstance()
if cmd.Operate == data_manager.ADDStream {
temp_stream_info := data_manager.NewStreamInfo(cmd.Name, cmd.DmcId, cmd.Status, cmd.Attributes, cmd.Type, list.New())
stream_instance.AddStreamInfo(cmd.Name, temp_stream_info)
} else if cmd.Operate == data_manager.DELStream {
stream_instance.DelStreamInfo(cmd.Name)
} else {
temp_stream_info := data_manager.NewStreamInfo(cmd.Name, cmd.DmcId, cmd.Status, cmd.Attributes, cmd.Type, list.New())
stream_instance.ModStreamInfo(cmd.Name, temp_stream_info)
}
gmcResponse := &GmcResponse{"gmc_response", cmd.MsgId}
buf, err := BuildGmcMessage(gmcResponse)
if err != nil {
return err
}
fmt.Println("------------- Session write frame: ---")
session.Write(buf)
return nil
}
func (dcr *DmcCommandStreamUpdate) GetCommandName() string {
return dcr.cmdAtt.CommandName
}
func (dcr *DmcCommandStreamUpdate) IsProvidedResult() bool {
return dcr.cmdAtt.IsProvidedResult
}
//////////////////// dmc_get_dms_request ////////////////
func (dcr *DmcCommandGetDmsRequest) CommandHandle(session *golis.Iosession, CmdInfo interface{}) error {
fmt.Println("get dms request command handle ........ ")
cmd := CmdInfo.(*DmcGetDmsRequest)
fmt.Println(" cmd:", cmd.Command, " msg id:", cmd.MsgId, " dmc id:", cmd.DmcId)
dmcManager := data_manager.GetDmcInstance()
dmc, err1 := dmcManager.GetDmc(data_manager.GET_A_USEFUL_DMS)
if err1 != nil {
return errors.New("source dmc session is nil")
}
session_source_dmc := dmc.Session
gmcGetDmsRequest := &GmcGetDmsRequest{"gmc_get_dms_request", cmd.MsgId}
buf, err := BuildGmcMessage(gmcGetDmsRequest)
if err != nil {
return err
}
fmt.Println("------------- Session write gmcGetDmsRequest frame: ---")
session_source_dmc.Write(buf)
return nil
}
func (dcr *DmcCommandGetDmsRequest) GetCommandName() string {
return dcr.cmdAtt.CommandName
}
func (dcr *DmcCommandGetDmsRequest) IsProvidedResult() bool {
return dcr.cmdAtt.IsProvidedResult
}
楼主
标题置顶
标题高亮
