onem2m-mgr.go

/*
 * Copyright (c) 2024-2025  The AdvantEDGE Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package sssmgr

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"reflect"
	"strconv"
	"strings"
	"sync"
	"time"

	log "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger"

	uuid "github.com/google/uuid"
)

// Sensors-Sharing Service Manager
type SssMgr struct {
	name                 string
	namespace            string
	bindingProtocol      string
	host                 string
	port                 int
	cse_name             string
	hostId               string
	mutex                sync.Mutex
	wg                   sync.WaitGroup
	refreshTicker        *time.Ticker
	sss_discovery_notify func()
	sss_status_notify    func()
	sss_data_notify      func()
}

type IotPlatformInfo struct {
	Address       string
	Port          int
	Name          string
	IotPlatformId string
}

type Point struct {
	Latitude  float64
	Longitude float64
}

type SensorCharacteristic struct {
	CharacteristicName          string
	CharacteristicValue         string
	CharacteristicUnitOfMeasure *string
}

type SensorDiscoveryInfo struct {
	SensorIdentifier         string
	SensorType               string
	SensorPropertyList       []string
	SensorCharacteristicList []SensorCharacteristic
	SensorPosition           *Point
	IotPlatformId            string
}

var registeredIotPlatformsMap = map[string]IotPlatformInfo{} // List of discovered IOT Plateform
var sensorsMap = map[string]SensorDiscoveryInfo{}            // Map sensors by sensorIdentifier
var sensorsPerPlatformMap = map[string][]string{}            // Map dsensorIdentifiers per platform

// Timer to refresh devices list for all IoT platform
const refreshTickerExpeary = 10 // In seconds

// Enable profiling
const profiling = false

var profilingTimers map[string]time.Time

var protocol SssMgrBindingProtocol

// NewSssMgr - Creates and initializes a new SSS Traffic Manager
func NewSssMgr(name string, namespace string, bindingProtocol string, host string, port int, hostId string, cse_name string, sss_discovery_notify func(), sss_status_notify func(), sss_data_notify func()) (tm *SssMgr, err error) {
	if name == "" {
		err = errors.New("Missing connector name")
		return nil, err
	}

	// Create new Traffic Manager
	tm = new(SssMgr)
	tm.name = name
	if namespace != "" {
		tm.namespace = namespace
	} else {
		tm.namespace = "default"
	}

	tm.bindingProtocol = bindingProtocol
	tm.host = host
	tm.port = port
	tm.cse_name = cse_name
	tm.hostId = hostId
	if tm.bindingProtocol == "MQTT" {
		if tm.host == "" {
			err := errors.New("Host not set for MQTTP protocol")
			log.Error(err.Error())
			return nil, err
		}
		if tm.port == 0 {
			tm.port = 1883
		}
		protocol = NewSssMgrMqtt()
	} else if tm.bindingProtocol == "REST_HTTP" {
		if tm.port == 0 {
			tm.port = 80
		}
		protocol = NewSssMgrHttp()
	} else {
		err := errors.New("Binding protocol not set")
		log.Error(err.Error())
		return nil, err
	}
	if hostId == "" {
		err := errors.New("hostId not set")
		log.Error(err.Error())
		return nil, err
	}
	if cse_name == "" {
		err := errors.New("cse_name not set")
		log.Error(err.Error())
		return nil, err
	}

	err = protocol.init(tm)
	if err != nil {
		log.Error(err.Error())
		return nil, err
	}

	tm.sss_discovery_notify = sss_discovery_notify
	tm.sss_status_notify = sss_status_notify
	tm.sss_data_notify = sss_data_notify

	tm.init()

	return tm, nil
}

// Profiling init
func (tm *SssMgr) init() {
	if profiling {
		profilingTimers = make(map[string]time.Time)
	}

	registeredIotPlatformsMap = make(map[string]IotPlatformInfo, 1)
	registeredIotPlatformsMap[tm.hostId] = IotPlatformInfo{
		Address:       tm.host,
		Port:          tm.port,
		Name:          tm.cse_name,
		IotPlatformId: tm.hostId,
	}
	sensorsMap = make(map[string]SensorDiscoveryInfo, 0)
	sensorsPerPlatformMap = make(map[string][]string, 0)
	tm.refreshTicker = nil

}

// DeleteSssMgr -
func (tm *SssMgr) DeleteSssMgr() (err error) {
	tm.stopRefreshTicker()

	if protocol != nil {
		protocol.uninit()
		protocol = nil
	}
	return nil
}

func (tm *SssMgr) startRefreshTicker() {
	log.Debug("Starting refresh loop")
	tm.refreshTicker = time.NewTicker(refreshTickerExpeary * time.Second)
	go func() {
		if tm.refreshTicker != nil {
			for range tm.refreshTicker.C {
				// Refresh the list of devices
				tm.wg.Add(1)
				err := tm.populateDevicesPerIotPlatforms()
				if err != nil {