Manage NamedQuery Consumers.

Author: Joerg Huber

SIF3InfraREST/SIF3REST/src/main/java/sif3/infra/rest/consumer/ConsumerLoader.java

@@ -71,6 +71,7 @@ public class ConsumerLoader
     private List<AbstractEventConsumer<?>> eventConsumers = new ArrayList<AbstractEventConsumer<?>>();
     private List<AbstractConsumer> crudConsumers = new ArrayList<AbstractConsumer>();
     private List<AbstractFunctionalServiceConsumer> fsServiceConsumers = new ArrayList<AbstractFunctionalServiceConsumer>();
+    private List<AbstractNamedQueryConsumer> nqConsumers = new ArrayList<AbstractNamedQueryConsumer>();
 
     private List<QueueReaderInfo<RemoteMessageQueueReader>> msgReaderServices = new ArrayList<QueueReaderInfo<RemoteMessageQueueReader>>();
 
@@ -233,6 +234,12 @@ private void shutdownConsumer()
             consumer.finalise();
         }
 
+        for (AbstractNamedQueryConsumer consumer : nqConsumers)
+        {
+            logger.debug("Shutdown " + consumer.getClass().getSimpleName());
+            consumer.finalise();
+        }
+
         if (getConsumerEnvironment().getEventsEnabled() || (getConsumerEnvironment().getDelayedEnabled()))
 		{
     		logger.debug("Shutdown Event Subscription Connector...");
@@ -307,6 +314,11 @@ private boolean initAsyncProcessor()
                 addServices(allLocalQueueCRUDServices, consumer.getAllApprovedCRUDServices(), consumer.getLocalConsumerQueue());
             }
 
+            for (AbstractNamedQueryConsumer consumer : nqConsumers)
+            {
+                addServices(allLocalQueueCRUDServices, consumer.getAllApprovedCRUDServices(), consumer.getLocalConsumerQueue());
+            }
+
             for (AbstractEventConsumer<?> consumer : eventConsumers)
             {
                 addServices(allLocalQueueCRUDServices, consumer.getAllApprovedCRUDServices(), consumer.getLocalConsumerQueue());
@@ -400,7 +412,7 @@ private QueueReaderInfo<RemoteMessageQueueReader> startRemoteMessageReaderThread
 		return queueReaderInfo;
 	}
 
-	private void initialiseConsumers(AdvancedProperties adapterProps)
+    private void initialiseConsumers(AdvancedProperties adapterProps)
 	{
 		List<String> classList = adapterProps.getFromCommaSeparated("consumer.classes");
 		String basePackageName = makePackageName(adapterProps.getPropertyAsString("consumer.basePackageName", ""));
@@ -417,6 +429,9 @@ private void initialiseConsumers(AdvancedProperties adapterProps)
 				// Instantiate class.
 				Object classObj = ct.newInstance();
 
+				// Init local queues. Must be called here as it is not part of the constructor.
+				((BaseConsumer)classObj).initLocalConsumerQueues();
+				
 				// Set properties and add it to correct structure
 				if (classObj instanceof AbstractEventConsumer)
 				{
@@ -433,9 +448,14 @@ else if (classObj instanceof AbstractFunctionalServiceConsumer)
                     logger.debug("Added " + classObj.getClass().getSimpleName() + " to fsServiceConsumers list");
                     fsServiceConsumers.add((AbstractFunctionalServiceConsumer) classObj);
 				}
+                else if (classObj instanceof AbstractNamedQueryConsumer)
+                {
+                    logger.debug("Added " + classObj.getClass().getSimpleName() + " to nqConsumers list");
+                    nqConsumers.add((AbstractNamedQueryConsumer) classObj);
+                }
 				else
 				{
-					logger.error("Consumer class " + className + " doesn't extend AbstractConsumer, AbstractEventConsumer of AbstractFunctionalServiceConsumer. Cannot initialse the Consumer.");
+					logger.error("Consumer class " + className + " doesn't extend AbstractConsumer, AbstractEventConsumer, AbstractFunctionalServiceConsumer or AbstractNamedQueryConsumer. Cannot initialse the Consumer.");
 				}
 			}
 			catch (Exception ex)

SIF3InfraREST/SIF3REST/src/main/java/sif3/infra/rest/queue/LocalMessageConsumer.java

@@ -306,15 +306,30 @@ private void processResponse(ResponseInfo responseInfo)
         	{
         	    Object payloadObject = null;
 
-                if (responseInfo.getServiceType() == ServiceType.FUNCTIONAL) 
+                switch (responseInfo.getServiceType())
                 {
-                    payloadObject = makeFunctionalServiceObject(responseInfo.getPayload(), responseInfo.getMediaType(), isPhaseObject, false);
-                }
-                else
-                {
-                    payloadObject = makeDataModelObject(minimalConsumer, responseInfo.getPayload(), responseInfo.getMediaType());
+                    case OBJECT: // no code here as it is the same as the SERVICEPATH below!
+                    case SERVICEPATH:
+                    {
+                        payloadObject = makeDataModelObject(minimalConsumer, responseInfo.getPayload(), responseInfo.getMediaType());
+                        break;
+                    }
+                    case FUNCTIONAL:
+                    {
+                        payloadObject = makeFunctionalServiceObject(responseInfo.getPayload(), responseInfo.getMediaType(), isPhaseObject, false);
+                        break;
+                    }
+                    case XQUERYTEMPLATE:
+                    {
+                        // Payload must remain a string. So there is no action required for this service type!
+                        break;
+                    }
+                    default:
+                    {
+                        logger.error("Received a Query Response for a Servic Type ("+responseInfo.getServiceType()+") that is not yet supported with this framework. Ignore message:\n"+responseInfo);
+                    }
                 }
-
+        	    
         	    if (payloadObject instanceof ErrorDetails)
         	    {
         	        // Something is not good at all. We send the error to the consumer.
@@ -336,6 +351,11 @@ private void processResponse(ResponseInfo responseInfo)
         			    delayedConsumer.onServicePath(payloadObject, new QueryCriteria(responseInfo.getUrlService()), paging, delayedReceipt);
         				break;
         			}
+                    case XQUERYTEMPLATE:
+                    {
+                        delayedConsumer.onQuery(payloadObject, paging, delayedReceipt);
+                        break;
+                    }
         			case FUNCTIONAL:
         			{
                         if (!isPhaseObject)

SIF3InfraREST/SIF3REST/src/main/java/sif3/infra/rest/queue/RemoteMessageQueueReader.java

@@ -339,7 +339,7 @@ private void processResponse(Response response)
 			{
 				logger.info("Received a DELAYED Response for which there is no consumer registered. Discard the following RESPONSE:\n" + response);
 			}
-			else // Create event object and send it to eventConsumer
+			else // Create message object and send it to blocking queue
 			{
 				logger.debug(getReaderID()+": Attempts to push DELAYED Response to local queue...");
 				localQueue.blockingPush(responseInfo);