pkg/control/duplicate.go

   1 /*
   2 ==================================================================================
   3   Copyright (c) 2021 Nokia
   4
   5    Licensed under the Apache License, Version 2.0 (the "License");
   6    you may not use this file except in compliance with the License.
   7    You may obtain a copy of the License at
   8
   9        http://www.apache.org/licenses/LICENSE-2.0
  10
  11    Unless required by applicable law or agreed to in writing, software
  12    distributed under the License is distributed on an "AS IS" BASIS,
  13    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14    See the License for the specific language governing permissions and
  15    limitations under the License.
  16 ==================================================================================
  17 */
  18
  19 package control
  20
  21 import (
  22         "bytes"
  23         "crypto/md5"
  24         "encoding/gob"
  25         "encoding/hex"
  26         "fmt"
  27         "sync"
  28         "time"
  29
  30         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  31 )
  32
  33 type retransEntry struct {
  34         restSubsId string
  35         startTime  time.Time
  36 }
  37
  38 type duplicateCtrl struct {
  39         mutex      sync.Mutex
  40         retransMap map[string]retransEntry
  41         collCount  int
  42 }
  43
  44 func (d *duplicateCtrl) Init() {
  45         d.retransMap = make(map[string]retransEntry)
  46 }
  47
  48 func (d *duplicateCtrl) IsDuplicateToOngoingTransaction(restSubsId string, payload interface{}) (error, bool, string) {
  49
  50         var data bytes.Buffer
  51         enc := gob.NewEncoder(&data)
  52
  53         if err := enc.Encode(payload); err != nil {
  54                 xapp.Logger.Error("Failed to encode %v\n", payload)
  55                 return err, false, ""
  56         }
  57
  58         hash := md5.Sum(data.Bytes())
  59
  60         md5sum := hex.EncodeToString(hash[:])
  61
  62         d.mutex.Lock()
  63         defer d.mutex.Unlock()
  64
  65         entry, present := d.retransMap[md5sum]
  66
  67         if present {
  68                 xapp.Logger.Info("Collision detected. REST subs ID %s has ongoing transaction with MD5SUM : %s started at %s\n", entry.restSubsId, md5sum, entry.startTime.Format(time.ANSIC))
  69                 d.collCount++
  70                 return nil, true, md5sum
  71         }
  72
  73         entry = retransEntry{restSubsId: restSubsId, startTime: time.Now()}
  74
  75         xapp.Logger.Debug("Added Md5SUM %s for restSubsId %s at %s\n", md5sum, entry.restSubsId, entry.startTime)
  76
  77         d.retransMap[md5sum] = entry
  78
  79         return nil, false, md5sum
  80 }
  81
  82 func (d *duplicateCtrl) TransactionComplete(md5sum string) error {
  83
  84         d.mutex.Lock()
  85         defer d.mutex.Unlock()
  86
  87         entry, present := d.retransMap[md5sum]
  88
  89         if !present {
  90                 xapp.Logger.Error("MD5SUM : %s NOT found from table (%v)\n", md5sum, entry)
  91                 return fmt.Errorf("Retransmission entry not found for MD5SUM %s", md5sum)
  92         }
  93
  94         xapp.Logger.Debug("Releasing transaction duplicate blocker for %s, MD5SUM : %s\n", entry.restSubsId, md5sum)
  95
  96         delete(d.retransMap, md5sum)
  97
  98         return nil
  99 }