Preserve the message order

Change-Id: I12c3034f4fceb869bba24c5c38546bcca087445d
Signed-off-by: Mohamed Abukar <abukar.mohamed@nokia.com>

diff --git a/pkg/xapp/restapi.go b/pkg/xapp/restapi.go
old mode 100644 (file)
new mode 100755 (executable)
index 291a874..1eab7b6
--- a/pkg/xapp/restapi.go
+++ b/pkg/xapp/restapi.go
@@ -53,7 +53,7 @@ func NewRouter() *Router {
 func (r *Router) serviceChecker(inner http.HandlerFunc) http.HandlerFunc {
        return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
                Logger.Info("restapi: method=%s url=%s", req.Method, req.URL.RequestURI())
-               if req.URL.RequestURI() == AliveURL || (Rmr.IsReady() && r.CheckStatus()) {
+               if req.URL.RequestURI() == AliveURL || r.CheckStatus() {
                        inner.ServeHTTP(w, req)
                } else {
                        respondWithJSON(w, http.StatusServiceUnavailable, nil)

diff --git a/pkg/xapp/rmr.go b/pkg/xapp/rmr.go
index 9a3116d..d310b3a 100755 (executable)
--- a/pkg/xapp/rmr.go
+++ b/pkg/xapp/rmr.go
@@ -155,11 +155,14 @@ func (m *RMRClient) Worker(taskName string, msgSize int) {
                }
                m.UpdateStatCounter("Received")
 
+               m.msgWg.Add(1)
                go m.parseMessage(rxBuffer)
+               m.msgWg.Wait()
        }
 }
 
 func (m *RMRClient) parseMessage(rxBuffer *C.rmr_mbuf_t) {
+       defer m.msgWg.Done()
        if len(m.consumers) == 0 {
                Logger.Info("rmrClient: No message handlers defined, message discarded!")
                return

diff --git a/pkg/xapp/types.go b/pkg/xapp/types.go
old mode 100644 (file)
new mode 100755 (executable)
index 2e1eb09..251dae5
--- a/pkg/xapp/types.go
+++ b/pkg/xapp/types.go
@@ -32,6 +32,7 @@ type RMRClient struct {
        context       unsafe.Pointer
        ready         int
        wg            sync.WaitGroup
+       msgWg         sync.WaitGroup
        mux           sync.Mutex
        stat          map[string]Counter
        consumers     []MessageConsumer