implement webSock functionality in meep-rnis (bd7a3875) · Commits · MEC - Multi-access Edge Computing / ETSI MEC Sandbox

go-apps/meep-rnis/server/rnis.go

+205 −35

Original line number	Original line	Diff line number	Diff line
	@@ -1609,7 +1609,11 @@ func checkCcNotificationRegisteredSubscriptions(appId string, assocId *Associate
	notif.SrcEcgi = &oldEcgi		notif.SrcEcgi = &oldEcgi
	notif.TrgEcgi = []Ecgi{newEcgi}		notif.TrgEcgi = []Ecgi{newEcgi}
	notif.AssociateId = append(notif.AssociateId, notifAssociateId)		notif.AssociateId = append(notif.AssociateId, notifAssociateId)
	notif.Links.Subscription.Href = hostUrl.String() + basePath + "subscriptions/" + subsIdStr		notif.Links = &CaReconfNotificationLinks{
			&LinkType{
			Href: hostUrl.String() + basePath + "subscriptions/" + subsIdStr,
			},
			}

	if subscription.CallbackReference != "" {		if subscription.CallbackReference != "" {
	sendCcNotification(subscription.CallbackReference, notif, false)		sendCcNotification(subscription.CallbackReference, notif, false)
	@@ -1688,10 +1692,19 @@ func checkReNotificationRegisteredSubscriptions(appId string, assocId *Associate
	notif.Ecgi = &newEcgi		notif.Ecgi = &newEcgi
	notif.ErabQosParameters = &erabQos		notif.ErabQosParameters = &erabQos
	notif.AssociateId = append(notif.AssociateId, notifAssociateId)		notif.AssociateId = append(notif.AssociateId, notifAssociateId)
	notif.Links.Subscription.Href = hostUrl.String() + basePath + "subscriptions/" + subsIdStr		notif.Links = &CaReconfNotificationLinks{
			&LinkType{
			Href: hostUrl.String() + basePath + "subscriptions/" + subsIdStr,
			},
			}

	sendReNotification(subscription.CallbackReference, notif)		if subscription.CallbackReference != "" {
			sendReNotification(subscription.CallbackReference, notif, false)
	log.Info("Rab_establishment Notification" + "(" + subsIdStr + ")")		log.Info("Rab_establishment Notification" + "(" + subsIdStr + ")")
			} else {
			sendReNotification(subscription.WebsockNotifConfig.WebsocketUri, notif, true)
			log.Info("Rab_establishment Notification" + "(" + subsIdStr + ")")
			}
	}		}
	}		}
	}		}
	@@ -1763,10 +1776,19 @@ func checkRrNotificationRegisteredSubscriptions(appId string, assocId *Associate
	notif.Ecgi = &oldEcgi		notif.Ecgi = &oldEcgi
	notif.ErabReleaseInfo = &erabRelInfo		notif.ErabReleaseInfo = &erabRelInfo
	notif.AssociateId = append(notif.AssociateId, notifAssociateId)		notif.AssociateId = append(notif.AssociateId, notifAssociateId)
	notif.Links.Subscription.Href = hostUrl.String() + basePath + "subscriptions/" + subsIdStr		notif.Links = &CaReconfNotificationLinks{
			&LinkType{
			Href: hostUrl.String() + basePath + "subscriptions/" + subsIdStr,
			},
			}

	sendRrNotification(subscription.CallbackReference, notif)		if subscription.CallbackReference != "" {
			sendRrNotification(subscription.CallbackReference, notif, false)
	log.Info("Rab_release Notification" + "(" + subsIdStr + ")")		log.Info("Rab_release Notification" + "(" + subsIdStr + ")")
			} else {
			sendRrNotification(subscription.WebsockNotifConfig.WebsocketUri, notif, true)
			log.Info("Rab_release Notification" + "(" + subsIdStr + ")")
			}
	}		}
	}		}
	}		}
	@@ -1887,12 +1909,21 @@ func checkMrNotificationRegisteredSubscriptions(key string, jsonInfo string, ext
	}		}
	}		}
	}		}
	notif.Links.Subscription.Href = hostUrl.String() + basePath + "subscriptions/" + subsIdStr		notif.Links = &CaReconfNotificationLinks{
			&LinkType{
			Href: hostUrl.String() + basePath + "subscriptions/" + subsIdStr,
			},
			}

	if parentMeasExists {		if parentMeasExists {
			if subscription.CallbackReference != "" {
	log.Info("Sending RNIS notification ", subscription.CallbackReference)		log.Info("Sending RNIS notification ", subscription.CallbackReference)
	callbackReference := subscription.CallbackReference		go sendMrNotification(subscription.CallbackReference, notif, false)
	go sendMrNotification(callbackReference, notif)
	log.Info("Meas_Rep_Ue Notification" + "(" + subsIdStr + ")")		log.Info("Meas_Rep_Ue Notification" + "(" + subsIdStr + ")")
			} else {
			go sendMrNotification(subscription.WebsockNotifConfig.WebsocketUri, notif, true)
			log.Info("Meas_Rep_Ue Notification" + "(" + subsIdStr + ")")
			}
	}		}
	}		}
	}		}
	@@ -2040,16 +2071,25 @@ func checkNrMrNotificationRegisteredSubscriptions(key string, jsonInfo string, e
	if report5GNeighborOnly {		if report5GNeighborOnly {
	notif.EutraNeighCellMeasInfo = nil		notif.EutraNeighCellMeasInfo = nil
	}		}
	notif.Links.Subscription.Href = hostUrl.String() + basePath + "subscriptions/" + subsIdStr		notif.Links = &CaReconfNotificationLinks{
			&LinkType{
			Href: hostUrl.String() + basePath + "subscriptions/" + subsIdStr,
			},
			}

	if parentMeasExists {		if parentMeasExists {
			if subscription.CallbackReference != "" {
	log.Info("Sending RNIS notification ", subscription.CallbackReference)		log.Info("Sending RNIS notification ", subscription.CallbackReference)
	callbackReference := subscription.CallbackReference		go sendNrMrNotification(subscription.CallbackReference, notif, false)
	go sendNrMrNotification(callbackReference, notif)		log.Info("Nr_Meas_Rep_Ue Notification" + "(" + subsIdStr + ")")
			} else {
			go sendNrMrNotification(subscription.WebsockNotifConfig.WebsocketUri, notif, true)
	log.Info("Nr_Meas_Rep_Ue Notification" + "(" + subsIdStr + ")")		log.Info("Nr_Meas_Rep_Ue Notification" + "(" + subsIdStr + ")")
	}		}
	}		}
	}		}
	}		}
			}
	return nil		return nil
	}		}

	@@ -2065,11 +2105,7 @@ func sendCcNotification(notifyUrl string, notification CellChangeNotification, i

	switch isWebsocket {		switch isWebsocket {
	case true:		case true:
	u := url.URL{		c, _, err = websocket.DefaultDialer.Dial(notifyUrl, nil)
	Host: notifyUrl,
	//Path: "/"
	}
	c, _, err = websocket.DefaultDialer.Dial(u.String(), nil)
	if err != nil {		if err != nil {
	log.Error(err.Error())		log.Error(err.Error())
	break		break
	@@ -2090,7 +2126,9 @@ func sendCcNotification(notifyUrl string, notification CellChangeNotification, i
	break		break
	}		}

	resp.StatusCode = http.StatusNoContent		resp = &http.Response{
			StatusCode: http.StatusNoContent,
			}
	case false:		case false:
	resp, err = http.Post(notifyUrl, "application/json", bytes.NewBuffer(jsonNotif))		resp, err = http.Post(notifyUrl, "application/json", bytes.NewBuffer(jsonNotif))
	_ = httpLog.LogTx(notifyUrl, "POST", string(jsonNotif), resp, startTime)		_ = httpLog.LogTx(notifyUrl, "POST", string(jsonNotif), resp, startTime)
	@@ -2107,79 +2145,211 @@ func sendCcNotification(notifyUrl string, notification CellChangeNotification, i
	defer resp.Body.Close()		defer resp.Body.Close()
	}		}

	func sendReNotification(notifyUrl string, notification RabEstNotification) {		func sendReNotification(notifyUrl string, notification RabEstNotification, isWebsocket bool) {
	startTime := time.Now()		startTime := time.Now()
	jsonNotif, err := json.Marshal(notification)		jsonNotif, err := json.Marshal(notification)
	if err != nil {		if err != nil {
	log.Error(err.Error())		log.Error(err.Error())
	}		}

	resp, err := http.Post(notifyUrl, "application/json", bytes.NewBuffer(jsonNotif))		var resp *http.Response
	duration := float64(time.Since(startTime).Microseconds()) / 1000.0		var c *websocket.Conn

			switch isWebsocket {
			case true:
			c, _, err = websocket.DefaultDialer.Dial(notifyUrl, nil)
			if err != nil {
			log.Error(err.Error())
			break
			}

			// send message
			err = c.WriteMessage(websocket.TextMessage, []byte(jsonNotif))
			if err != nil {
			// handle error
			log.Error(err.Error())
			break
			}

			// receive message
			_, _, err = c.ReadMessage()
			if err != nil {
			log.Error(err.Error())
			break
			}

			resp = &http.Response{
			StatusCode: http.StatusNoContent,
			}
			case false:
			resp, err = http.Post(notifyUrl, "application/json", bytes.NewBuffer(jsonNotif))
	_ = httpLog.LogTx(notifyUrl, "POST", string(jsonNotif), resp, startTime)		_ = httpLog.LogTx(notifyUrl, "POST", string(jsonNotif), resp, startTime)
			}

			duration := float64(time.Since(startTime).Microseconds()) / 1000.0
	if err != nil {		if err != nil {
	log.Error(err)		log.Error(err)
	met.ObserveNotification(sandboxName, serviceName, notifRabEst, notifyUrl, nil, duration)		met.ObserveNotification(sandboxName, serviceName, notifRabEst, notifyUrl, nil, duration)
	return		return
	}		}
	met.ObserveNotification(sandboxName, serviceName, notifRabEst, notifyUrl, resp, duration)		met.ObserveNotification(sandboxName, serviceName, notifRabEst, notifyUrl, resp, duration)
			defer c.Close()
	defer resp.Body.Close()		defer resp.Body.Close()
	}		}

	func sendRrNotification(notifyUrl string, notification RabRelNotification) {		func sendRrNotification(notifyUrl string, notification RabRelNotification, isWebsocket bool) {
	startTime := time.Now()		startTime := time.Now()
	jsonNotif, err := json.Marshal(notification)		jsonNotif, err := json.Marshal(notification)
	if err != nil {		if err != nil {
	log.Error(err.Error())		log.Error(err.Error())
	}		}

	resp, err := http.Post(notifyUrl, "application/json", bytes.NewBuffer(jsonNotif))		var resp *http.Response
	duration := float64(time.Since(startTime).Microseconds()) / 1000.0		var c *websocket.Conn

			switch isWebsocket {
			case true:
			c, _, err = websocket.DefaultDialer.Dial(notifyUrl, nil)
			if err != nil {
			log.Error(err.Error())
			break
			}

			// send message
			err = c.WriteMessage(websocket.TextMessage, []byte(jsonNotif))
			if err != nil {
			// handle error
			log.Error(err.Error())
			break
			}

			// receive message
			_, _, err = c.ReadMessage()
			if err != nil {
			log.Error(err.Error())
			break
			}

			resp = &http.Response{
			StatusCode: http.StatusNoContent,
			}
			case false:
			resp, err = http.Post(notifyUrl, "application/json", bytes.NewBuffer(jsonNotif))
	_ = httpLog.LogTx(notifyUrl, "POST", string(jsonNotif), resp, startTime)		_ = httpLog.LogTx(notifyUrl, "POST", string(jsonNotif), resp, startTime)
			}

			duration := float64(time.Since(startTime).Microseconds()) / 1000.0
	if err != nil {		if err != nil {
	log.Error(err)		log.Error(err)
	met.ObserveNotification(sandboxName, serviceName, notifRabRel, notifyUrl, nil, duration)		met.ObserveNotification(sandboxName, serviceName, notifRabRel, notifyUrl, nil, duration)
	return		return
	}		}
	met.ObserveNotification(sandboxName, serviceName, notifRabRel, notifyUrl, resp, duration)		met.ObserveNotification(sandboxName, serviceName, notifRabRel, notifyUrl, resp, duration)
			defer c.Close()
	defer resp.Body.Close()		defer resp.Body.Close()
	}		}

	func sendMrNotification(notifyUrl string, notification MeasRepUeNotification) {		func sendMrNotification(notifyUrl string, notification MeasRepUeNotification, isWebsocket bool) {
	startTime := time.Now()		startTime := time.Now()
	jsonNotif, err := json.Marshal(notification)		jsonNotif, err := json.Marshal(notification)
	if err != nil {		if err != nil {
	log.Error(err.Error())		log.Error(err.Error())
	}		}

	resp, err := http.Post(notifyUrl, "application/json", bytes.NewBuffer(jsonNotif))		var resp *http.Response
	duration := float64(time.Since(startTime).Microseconds()) / 1000.0		var c *websocket.Conn

			switch isWebsocket {
			case true:
			c, _, err = websocket.DefaultDialer.Dial(notifyUrl, nil)
			if err != nil {
			log.Error(err.Error())
			break
			}

			// send message
			err = c.WriteMessage(websocket.TextMessage, []byte(jsonNotif))
			if err != nil {
			// handle error
			log.Error(err.Error())
			break
			}

			// receive message
			_, _, err = c.ReadMessage()
			if err != nil {
			log.Error(err.Error())
			break
			}

			resp = &http.Response{
			StatusCode: http.StatusNoContent,
			}
			case false:
			resp, err = http.Post(notifyUrl, "application/json", bytes.NewBuffer(jsonNotif))
	_ = httpLog.LogTx(notifyUrl, "POST", string(jsonNotif), resp, startTime)		_ = httpLog.LogTx(notifyUrl, "POST", string(jsonNotif), resp, startTime)
			}

			duration := float64(time.Since(startTime).Microseconds()) / 1000.0
	if err != nil {		if err != nil {
	log.Error(err)		log.Error(err)
	met.ObserveNotification(sandboxName, serviceName, notifMeasRepUe, notifyUrl, nil, duration)		met.ObserveNotification(sandboxName, serviceName, notifMeasRepUe, notifyUrl, nil, duration)
	return		return
	}		}
	met.ObserveNotification(sandboxName, serviceName, notifMeasRepUe, notifyUrl, resp, duration)		met.ObserveNotification(sandboxName, serviceName, notifMeasRepUe, notifyUrl, resp, duration)
			defer c.Close()
	defer resp.Body.Close()		defer resp.Body.Close()
	}		}

	func sendNrMrNotification(notifyUrl string, notification NrMeasRepUeNotification) {		func sendNrMrNotification(notifyUrl string, notification NrMeasRepUeNotification, isWebsocket bool) {
	startTime := time.Now()		startTime := time.Now()
	jsonNotif, err := json.Marshal(notification)		jsonNotif, err := json.Marshal(notification)
	if err != nil {		if err != nil {
	log.Error(err.Error())		log.Error(err.Error())
	}		}

	resp, err := http.Post(notifyUrl, "application/json", bytes.NewBuffer(jsonNotif))		var resp *http.Response
	duration := float64(time.Since(startTime).Microseconds()) / 1000.0		var c *websocket.Conn

			switch isWebsocket {
			case true:
			c, _, err = websocket.DefaultDialer.Dial(notifyUrl, nil)
			if err != nil {
			log.Error(err.Error())
			break
			}

			// send message
			err = c.WriteMessage(websocket.TextMessage, []byte(jsonNotif))
			if err != nil {
			// handle error
			log.Error(err.Error())
			break
			}

			// receive message
			_, _, err = c.ReadMessage()
			if err != nil {
			log.Error(err.Error())
			break
			}

			resp = &http.Response{
			StatusCode: http.StatusNoContent,
			}
			case false:
			resp, err = http.Post(notifyUrl, "application/json", bytes.NewBuffer(jsonNotif))
	_ = httpLog.LogTx(notifyUrl, "POST", string(jsonNotif), resp, startTime)		_ = httpLog.LogTx(notifyUrl, "POST", string(jsonNotif), resp, startTime)
			}

			duration := float64(time.Since(startTime).Microseconds()) / 1000.0
	if err != nil {		if err != nil {
	log.Error(err)		log.Error(err)
	met.ObserveNotification(sandboxName, serviceName, notifNrMeasRepUe, notifyUrl, nil, duration)		met.ObserveNotification(sandboxName, serviceName, notifNrMeasRepUe, notifyUrl, nil, duration)
	return		return
	}		}
	met.ObserveNotification(sandboxName, serviceName, notifNrMeasRepUe, notifyUrl, resp, duration)		met.ObserveNotification(sandboxName, serviceName, notifNrMeasRepUe, notifyUrl, resp, duration)
			defer c.Close()
	defer resp.Body.Close()		defer resp.Body.Close()
	}		}