Merge pull request #1486 from FIWARE/aeros/httpHeader

Aeros/http header

src/lib/orionld/common/orionldState.h

@@ -246,6 +246,7 @@ typedef struct OrionldStateIn
   char*     tenant;
   char*     legacy;          // Use legacy mongodb driver / mongoBackend
   bool      performance;
+  bool      aerOS;           // Special treatment for aerOS specific features
 
   // Incoming payload
   char*     payload;

src/lib/orionld/forwarding/distOpRequests.cpp

@@ -138,7 +138,7 @@ DistOp* distOpRequests(char* entityId, char* entityType, DistOpType operation, K
     // Send the forwarded request and await all responses
     if ((distOpP->regP != NULL) && (distOpP->error == false))
     {
-      if (distOpSend(distOpP, dateHeader, xff) == 0)
+      if (distOpSend(distOpP, dateHeader, xff, false) == 0)
       {
         distOpP->error = false;
         orionldState.distOp.requests += 1;

src/lib/orionld/forwarding/distOpSend.cpp

@@ -325,7 +325,7 @@ void bodyCompact(DistOpType operation, KjNode* requestBody, OrionldContext* fwdC
 //
 // distOpSend -
 //
-bool distOpSend(DistOp* distOpP, const char* dateHeader, const char* xForwardedForHeader)
+bool distOpSend(DistOp* distOpP, const char* dateHeader, const char* xForwardedForHeader, bool local)
 {
   //
   // Figure out the @context to use for the forwarded request
@@ -408,6 +408,9 @@ bool distOpSend(DistOp* distOpP, const char* dateHeader, const char* xForwardedF
     if ((distOpP->operation == DoQueryEntity) && (distOpP->entityId != NULL))
       uriParamAdd(&urlParts, "id", distOpP->entityId, -1);
 
+    if (local == true)
+      uriParamAdd(&urlParts, "local=true", NULL, 10);
+
     //
     // If we know the Entity Type, we pass that piece of information as well
     //
@@ -427,7 +430,7 @@ bool distOpSend(DistOp* distOpP, const char* dateHeader, const char* xForwardedF
   if (orionldState.uriParams.qCopy != NULL)
   {
     uriParamAdd(&urlParts, "q", orionldState.uriParams.qCopy, -1);
-    LM_T(LmtDistOpRequest, ("%s: orionldState.uriParams.q: '%s'", distOpP->regP->regId, orionldState.uriParams.qCopy));
+    LM_T(LmtDistOpRequestHeaders, ("%s: orionldState.uriParams.q: '%s'", distOpP->regP->regId, orionldState.uriParams.qCopy));
   }
 
   //

src/lib/orionld/forwarding/distOpSend.h

@@ -33,6 +33,6 @@
 //
 // distOpSend -
 //
-extern bool distOpSend(DistOp* distOpP, const char* dateHeader, const char* xForwardedForHeader);
+extern bool distOpSend(DistOp* distOpP, const char* dateHeader, const char* xForwardedForHeader, bool local);
 
 #endif  // SRC_LIB_ORIONLD_FORWARDING_DISTOPSEND_H_

src/lib/orionld/forwarding/distOpsSend.cpp

@@ -36,7 +36,7 @@
 //
 // distOpsSend -
 //
-int distOpsSend(DistOp* distOpList)
+int distOpsSend(DistOp* distOpList, bool local)
 {
   char* xff = xForwardedForCompose(orionldState.in.xForwardedFor, localIpAndPort);
 
@@ -51,7 +51,7 @@ int distOpsSend(DistOp* distOpList)
     {
       distOpP->onlyIds = true;
 
-      if (distOpSend(distOpP, dateHeader, xff) == 0)
+      if (distOpSend(distOpP, dateHeader, xff, local) == 0)
         distOpP->error = false;
       else
         distOpP->error = true;
@@ -118,7 +118,7 @@ int distOpsSend2(DistOpListItem* distOpList)
     {
       distOpP->onlyIds = false;
 
-      if (distOpSend(distOpP, dateHeader, xff) == 0)
+      if (distOpSend(distOpP, dateHeader, xff, false) == 0)
         distOpP->error = false;
       else
         distOpP->error = true;

src/lib/orionld/forwarding/distOpsSend.h

@@ -33,7 +33,7 @@
 //
 // distOpsSend -
 //
-extern int distOpsSend(DistOp* distOpList);
+extern int distOpsSend(DistOp* distOpList, bool local);
 
 
 

src/lib/orionld/rest/orionldMhdConnectionInit.cpp

@@ -433,6 +433,11 @@ static MHD_Result orionldHttpHeaderReceive(void* cbDataP, MHD_ValueKind kind, co
   }
   else if (strcasecmp(key, "Performance") == 0)
     orionldState.in.performance = true;
+  else if (strcasecmp(key, "aerOS") == 0)
+  {
+    if (strcasecmp(value, "true") == 0)
+      orionldState.in.aerOS = true;
+  }
   else if (strcasecmp(key, "NGSILD-Scope") == 0)
   {
     orionldState.scopes = strSplit((char*) value, ',', orionldState.scopeV, K_VEC_SIZE(orionldState.scopeV));

src/lib/orionld/serviceRoutines/orionldDeleteAttribute.cpp

@@ -95,7 +95,7 @@ static DistOp* distributedDelete(KjNode* responseBody, char* entityId, char* ent
       char dateHeader[70];
       snprintf(dateHeader, sizeof(dateHeader), "Date: %s", orionldState.requestTimeString);
 
-      if (distOpSend(distOpP, dateHeader, xff) == 0)
+      if (distOpSend(distOpP, dateHeader, xff, false) == 0)
       {
         ++forwards;
         distOpP->error = false;

src/lib/orionld/serviceRoutines/orionldDeleteEntity.cpp

@@ -122,7 +122,7 @@ static DistOp* distributedDelete(char* entityId, char* entityTypeExpanded, char*
       char dateHeader[70];
       snprintf(dateHeader, sizeof(dateHeader), "Date: %s", orionldState.requestTimeString);
 
-      if (distOpSend(distOpP, dateHeader, xff) == 0)
+      if (distOpSend(distOpP, dateHeader, xff, false) == 0)
       {
         ++forwards;
         distOpP->error = false;

src/lib/orionld/serviceRoutines/orionldGetEntitiesDistributed.cpp

@@ -281,7 +281,7 @@ static void distOpMatchIdsRequest(DistOp* distOpList)
 
   distOpListDebug2(distOpList, "DistOps before sending the onlyId=true requests");
   // Send all distributed requests
-  int forwards = distOpsSend(distOpList);
+  int forwards = distOpsSend(distOpList, orionldState.in.aerOS);
 
   // Await all responses, if any
   if (forwards > 0)

src/lib/orionld/serviceRoutines/orionldGetEntitiesPage.cpp

@@ -125,12 +125,20 @@ DistOpListItem* distOpListItemCreate(const char* distOpId, char* idString)
 
   if (distOpP == NULL)
   {
+#if 0
     //
     // I think this LM_RE here is incorrect.
     // The DistOps are for one request only.
     // So, create a new DistOp if it does not exist.
     //
     LM_RE(NULL, ("Internal Error (unable to find the DistOp '%s'", distOpId));
+#else
+    RegCacheItem* rciP = regCacheItemLookup(orionldState.tenantP->regCache, distOpId);
+    distOpP = distOpCreate(DoQueryEntity, rciP, NULL, NULL, NULL);
+    // Add DistOp to the linked list of DistOps
+    distOpP->next    = orionldState.distOpList;
+    orionldState.distOpList = distOpP;
+#endif
   }
 
   DistOpListItem* itemP = (DistOpListItem*) kaAlloc(&orionldState.kalloc, sizeof(DistOpListItem));
@@ -169,18 +177,12 @@ DistOpListItem* distOpListItemAdd(DistOpListItem* distOpList, const char* distOp
 
 // -----------------------------------------------------------------------------
 //
-// queryResponse -
-//
-// FIXME:
-//   I have a problem here with responses for requests forwarded due to Auxiliary Registrations.
-//   They can't be taken into account until AFTER all other responses have been dealt with,
-//   So, unless all other responses have been treated (merging entities), responses from Aux Regs must be delayed.
-//   They must be kept in a list and treated once no other responses are left.
+// responseMerge -
 //
-static int queryResponse(DistOp* distOpP, void* callbackParam)
+static void responseMerge(DistOp* distOpP, KjNode* entityArray)
 {
-  LM_T(LmtSR, ("Got a response. status code: %d. callbackParam: %p", distOpP->httpResponseCode, callbackParam));
-  KjNode* entityArray = (KjNode*) callbackParam;
+  LM_W(("Merging entities for DistOp '%s' (aux: %s)", distOpP->id, (distOpP->regP->mode == RegModeAuxiliary)? "YES" : "NO"));
+  LM_T(LmtSR, ("Got a response. status code: %d. entityArray: %p", distOpP->httpResponseCode, entityArray));
 
   kjTreeLog(distOpP->responseBody, "Response", LmtSR);
 
@@ -214,9 +216,8 @@ static int queryResponse(DistOp* distOpP, void* callbackParam)
         }
         else
         {
-          bool auxiliary = distOpP->regP->mode == RegModeAuxiliary;
           LM_T(LmtDistOpMerge, ("Existing Entity '%s' - merging it in the entity array (reg-mode: %s)", entityId, registrationModeToString(distOpP->regP->mode)));
-          distOpEntityMerge(baseEntityP, entityP, orionldState.uriParamOptions.sysAttrs, auxiliary);
+          distOpEntityMerge(baseEntityP, entityP, orionldState.uriParamOptions.sysAttrs, distOpP->regP->mode == RegModeAuxiliary);
         }
       }
       else
@@ -225,7 +226,19 @@ static int queryResponse(DistOp* distOpP, void* callbackParam)
       entityP = next;
     }
   }
+}
+
+
 
+// -----------------------------------------------------------------------------
+//
+// queryResponse -
+//
+static int queryResponse(DistOp* distOpP, void* callbackParam)
+{
+  KjNode* entityArray = (KjNode*) callbackParam;
+
+  responseMerge(distOpP, entityArray);
   return 0;
 }
 
@@ -243,7 +256,7 @@ typedef int (*DistOpResponseTreatFunction)(DistOp* distOpP, void* callbackParam)
 //
 // distOpsReceive2 - FIXME: move to orionld/forwarding/distOpsReceive.cpp/h
 //
-void distOpsReceive2(DistOpResponseTreatFunction treatFunction, void* callbackParam)
+void distOpsReceive2(DistOpListItem* distOpList, DistOpResponseTreatFunction treatFunction, void* callbackParam)
 {
   LM_T(LmtSR, ("Receiving responses"));
   //
@@ -275,9 +288,18 @@ void distOpsReceive2(DistOpResponseTreatFunction treatFunction, void* callbackPa
       LM_T(LmtDistOpResponse, ("%s: received a response for a forwarded request", distOpP->regP->regId, distOpP->httpResponseCode));
       LM_T(LmtDistOpResponse, ("%s: response for a forwarded request: %s", distOpP->regP->regId, distOpP->rawResponse));
 
-      treatFunction(distOpP, callbackParam);
+      if (distOpP->regP->mode == RegModeAuxiliary)
+        LM_W(("%s: Aux registration - merge is delayed", distOpP->regP->regId));
+      else
+        treatFunction(distOpP, callbackParam);
     }
   }
+
+  for (DistOp* distOpP = orionldState.distOpList; distOpP != NULL; distOpP = distOpP->next)
+  {
+    if ((distOpP->regP != NULL) && (distOpP->regP->mode == RegModeAuxiliary))
+      responseMerge(distOpP, (KjNode*) callbackParam);
+  }
 }
 
 
@@ -294,7 +316,7 @@ static void distOpQueryRequest(DistOpListItem* distOpList, KjNode* entityArray)
 
   // Await all responses, if any
   if (forwards > 0)
-    distOpsReceive2(queryResponse, entityArray);
+    distOpsReceive2(distOpList, queryResponse, entityArray);
 }
 
 

src/lib/orionld/serviceRoutines/orionldGetEntity.cpp

@@ -190,7 +190,7 @@ bool orionldGetEntity(void)
         if (distOpP->regP != NULL)
         {
           LM_T(LmtDistOpAttributes, ("distOp::attrsParam: '%s'", distOpP->attrsParam));
-          if (distOpSend(distOpP, dateHeader, xff) == 0)
+          if (distOpSend(distOpP, dateHeader, xff, false) == 0)
           {
             ++forwards;
             distOpP->error = false;

src/lib/orionld/serviceRoutines/orionldPostBatchDelete.cpp

@@ -478,7 +478,7 @@ bool orionldPostBatchDelete(void)
           char dateHeader[70];
           snprintf(dateHeader, sizeof(dateHeader), "Date: %s", orionldState.requestTimeString);
 
-          if (distOpSend(distReqP, dateHeader, xff) == 0)
+          if (distOpSend(distReqP, dateHeader, xff, false) == 0)
           {
             ++forwards;
             distReqP->error = false;

test/functionalTest/cases/0000_ngsild/ngsild_entityMap_crossed_registrations_multiple_forwarding.test

@@ -272,7 +272,7 @@ echo
 
 echo "13. Create an entity map for a GET /entities?type=T in CB"
 echo "========================================================="
-orionCurl --url '/ngsi-ld/v1/entities?type=T&count=true'
+orionCurl --url '/ngsi-ld/v1/entities?type=T&count=true' -H "aerOS: true" 
 entityMap=$(echo "$_responseHeaders" | grep Entity-Map: | awk -F ': ' '{ print $2 }' | tr -d "\r\n")
 echo
 echo
@@ -287,7 +287,7 @@ echo
 
 echo "15. Create an entity map for a GET /entities?type=T in CP1"
 echo "=========================================================="
-orionCurl --url '/ngsi-ld/v1/entities?type=T&count=true&reset=true' --port $CP1_PORT
+orionCurl --url '/ngsi-ld/v1/entities?type=T&count=true&reset=true'  -H "aerOS: true" --port $CP1_PORT
 entityMap=$(echo "$_responseHeaders" | grep Entity-Map: | awk -F ': ' '{ print $2 }' | tr -d "\r\n")
 echo
 echo
@@ -302,7 +302,7 @@ echo
 
 echo "17. Create an entity map for a GET /entities?type=T in CP2"
 echo "=========================================================="
-orionCurl --url '/ngsi-ld/v1/entities?type=T&count=true&reset=true' --port $CP2_PORT
+orionCurl --url '/ngsi-ld/v1/entities?type=T&count=true&reset=true' -H "aerOS: true" --port $CP2_PORT
 entityMap=$(echo "$_responseHeaders" | grep Entity-Map: | awk -F ': ' '{ print $2 }' | tr -d "\r\n")
 echo
 echo
@@ -315,7 +315,6 @@ echo
 echo
 
 
-
 --REGEXPECT--
 01. In CB, create an entity urn:E1 with attributes P1+R1
 ========================================================
@@ -655,11 +654,17 @@ Date: REGEX(.*)
         "@none"
     ],
     "urn:E5": [
+<<<<<<< HEAD
+        "urn:R2"
+    ],
+    "urn:E6": [
+=======
         "urn:R1",
         "urn:R2"
     ],
     "urn:E6": [
         "urn:R1",
+>>>>>>> feature/entityMaps
         "urn:R2"
     ]
 }
@@ -762,6 +767,17 @@ Date: REGEX(.*)
 
 {
     "urn:E1": [
+<<<<<<< HEAD
+        "urn:R1"
+    ],
+    "urn:E2": [
+        "urn:R1"
+    ],
+    "urn:E3": [
+        "urn:R2"
+    ],
+    "urn:E4": [
+=======
         "urn:R1",
         "urn:R2"
     ],
@@ -775,6 +791,7 @@ Date: REGEX(.*)
     ],
     "urn:E4": [
         "urn:R1",
+>>>>>>> feature/entityMaps
         "urn:R2"
     ],
     "urn:E5": [