only that set of message which are specified in limit.\r
\r
.. code:: bash\r
+ \r
Suppose if we set limit=2, then only 2 sets of data will be consumed. \r
*Get \ **<<topicName>>/group/2?limit=4*** \r
Let us suppose if \r
\r
Remember, only Highland Park standard library filter components can be\r
used -- no plug-ins are available in the Cambria server context.\r
+ \r
+ .. code:: bash\r
+ \r
+ package org.onap.sa.highlandPark.integration;\r
\r
-package org.onap.sa.highlandPark.integration;\r
+ import java.io.IOException;\r
\r
-import java.io.IOException;\r
+ import java.util.UUID;\r
\r
-import java.util.UUID;\r
+ import org.onap.nsa.cambria.client.CambriaClientFactory;\r
\r
-import org.onap.nsa.cambria.client.CambriaClientFactory;\r
+ import org.onap.nsa.cambria.client.CambriaConsumer;\r
\r
-import org.onap.nsa.cambria.client.CambriaConsumer;\r
+ import org.onap.sa.highlandPark.processor.HpEvent;\r
\r
-import org.onap.sa.highlandPark.processor.HpEvent;\r
+ import org.onap.sa.highlandPark.stdlib.filters.FilterIo;\r
\r
-import org.onap.sa.highlandPark.stdlib.filters.FilterIo;\r
+ import org.onap.sa.highlandPark.stdlib.filters.OneOf;\r
\r
-import org.onap.sa.highlandPark.stdlib.filters.OneOf;\r
+ public class ExampleFilteringConsumer\r
\r
-public class ExampleFilteringConsumer\r
+ {\r
\r
-{\r
+ public static void main ( String[] args ) throws IOException,\r
+ InterruptedException\r
\r
-public static void main ( String[] args ) throws IOException,\r
-InterruptedException\r
+ {\r
\r
-{\r
+ // Cambria clients take a set of 1 or more servers to use in round-robin\r
+ fashion.\r
\r
-// Cambria clients take a set of 1 or more servers to use in round-robin\r
-fashion.\r
+ // If a server becomes unreachable, another in the group is used.\r
\r
-// If a server becomes unreachable, another in the group is used.\r
+ final String\r
+ serverGroup="ueb01hydc.it.att.com,ueb02hydc.it.att.com,ueb03hydc.it.att.com";\r
\r
-final String\r
-serverGroup="ueb01hydc.it.att.com,ueb02hydc.it.att.com,ueb03hydc.it.att.com";\r
+ // choose a topic\r
\r
-// choose a topic\r
+ final String topic = "TEST-TOPIC";\r
\r
-final String topic = "TEST-TOPIC";\r
+ // Cambria clients can run in a cooperative group to handle high-volume\r
+ topics.\r
\r
-// Cambria clients can run in a cooperative group to handle high-volume\r
-topics.\r
+ // Here, we create a random group name, which means this client is not\r
+ re-startable.\r
\r
-// Here, we create a random group name, which means this client is not\r
-re-startable.\r
+ final String consumerGroup = UUID.randomUUID ().toString ();\r
\r
-final String consumerGroup = UUID.randomUUID ().toString ();\r
+ final String consumerId = "0";\r
\r
-final String consumerId = "0";\r
+ // Cambria clients can sit in a tight loop on the client side, using a\r
+ long-poll\r
\r
-// Cambria clients can sit in a tight loop on the client side, using a\r
-long-poll\r
+ // to wait for messages, and a limit to tell the server the most to send\r
+ at a time.\r
\r
-// to wait for messages, and a limit to tell the server the most to send\r
-at a time.\r
+ final int longPollMs = 30\*1000;\r
\r
-final int longPollMs = 30\*1000;\r
+ final int limit = -1;\r
\r
-final int limit = -1;\r
+ // The Cambria server can filter the returned message stream using\r
+ filters from the\r
\r
-// The Cambria server can filter the returned message stream using\r
-filters from the\r
+ // Highland Park system. Here, we create a simple filter to test for the\r
+ AlarmID\r
\r
-// Highland Park system. Here, we create a simple filter to test for the\r
-AlarmID\r
+ // value being one of the Mobility power alarms.\r
\r
-// value being one of the Mobility power alarms.\r
+ final OneOf oneOf = new OneOf ( "AlarmId", kPowerAlarms );\r
\r
-final OneOf oneOf = new OneOf ( "AlarmId", kPowerAlarms );\r
+ // create the consumer\r
\r
-// create the consumer\r
+ final CambriaConsumer cc = CambriaClientFactory.createConsumer (\r
+ serverGroup, topic,\r
\r
-final CambriaConsumer cc = CambriaClientFactory.createConsumer (\r
-serverGroup, topic,\r
+ consumerGroup, consumerId, longPollMs, limit, FilterIo.write ( oneOf )\r
+ );\r
\r
-consumerGroup, consumerId, longPollMs, limit, FilterIo.write ( oneOf )\r
-);\r
+ // now loop reading messages. Note that cc.fetch() will wait in its HTTP\r
+ receive\r
\r
-// now loop reading messages. Note that cc.fetch() will wait in its HTTP\r
-receive\r
+ // method for up to 30 seconds (longPollMs) when nothing's available at\r
+ the server.\r
\r
-// method for up to 30 seconds (longPollMs) when nothing's available at\r
-the server.\r
+ long count = 0;\r
\r
-long count = 0;\r
+ while ( true )\r
\r
-while ( true )\r
+ {\r
\r
-{\r
+ for ( String msg : cc.fetch () )\r
\r
-for ( String msg : cc.fetch () )\r
+ {\r
\r
-{\r
+ System.out.println ( "" + (++count) + ": " + msg );\r
\r
-System.out.println ( "" + (++count) + ": " + msg );\r
+ }\r
\r
-}\r
+ }\r
\r
-}\r
+ }\r
\r
-}\r
+ private static final String[] kPowerAlarms =\r
\r
-private static final String[] kPowerAlarms =\r
+ {\r
\r
-{\r
+ "HUB COMMERCIAL POWER FAIL\_FWD",\r
\r
-"HUB COMMERCIAL POWER FAIL\_FWD",\r
+ "HUB COMMERCIAL POWER FAIL",\r
\r
-"HUB COMMERCIAL POWER FAIL",\r
+ "RBS COMMERCIAL POWER FAIL - Fixed\_FWD",\r
\r
-"RBS COMMERCIAL POWER FAIL - Fixed\_FWD",\r
+ "RBS COMMERCIAL POWER FAIL\_FWD",\r
\r
-"RBS COMMERCIAL POWER FAIL\_FWD",\r
+ "RBS COMMERCIAL POWER FAIL - No Generator\_FWD",\r
\r
-"RBS COMMERCIAL POWER FAIL - No Generator\_FWD",\r
+ "RBS COMMERCIAL POWER FAIL - Portable\_FWD",\r
\r
-"RBS COMMERCIAL POWER FAIL - Portable\_FWD",\r
+ "RBS COMMERCIAL POWER FAIL - Shared\_FWD",\r
\r
-"RBS COMMERCIAL POWER FAIL - Shared\_FWD",\r
+ "RBS COMMERCIAL POWER FAIL - Yes\_FWD",\r
\r
-"RBS COMMERCIAL POWER FAIL - Yes\_FWD",\r
+ "RBS COMMERCIAL POWER FAIL - YES\_FWD",\r
\r
-"RBS COMMERCIAL POWER FAIL - YES\_FWD",\r
+ "RBS COMMERCIAL POWER FAIL - Fixed",\r
\r
-"RBS COMMERCIAL POWER FAIL - Fixed",\r
+ "RBS COMMERCIAL POWER FAIL - No Generator",\r
\r
-"RBS COMMERCIAL POWER FAIL - No Generator",\r
+ "RBS COMMERCIAL POWER FAIL - Portable",\r
\r
-"RBS COMMERCIAL POWER FAIL - Portable",\r
+ "RBS COMMERCIAL POWER FAIL - Shared",\r
\r
-"RBS COMMERCIAL POWER FAIL - Shared",\r
+ "RBS COMMERCIAL POWER FAIL - YES",\r
\r
-"RBS COMMERCIAL POWER FAIL - YES",\r
+ "RBS COMMERCIAL POWER FAIL - Yes",\r
\r
-"RBS COMMERCIAL POWER FAIL - Yes",\r
+ "RBS COMMERCIAL POWER FAIL",\r
\r
-"RBS COMMERCIAL POWER FAIL",\r
+ "HUB COMMERCIAL POWER FAIL - Fixed",\r
\r
-"HUB COMMERCIAL POWER FAIL - Fixed",\r
+ "HUB COMMERCIAL POWER FAIL - No Generator",\r
\r
-"HUB COMMERCIAL POWER FAIL - No Generator",\r
+ "HUB COMMERCIAL POWER FAIL - Portable",\r
\r
-"HUB COMMERCIAL POWER FAIL - Portable",\r
+ "HUB COMMERCIAL POWER FAIL - Shared",\r
\r
-"HUB COMMERCIAL POWER FAIL - Shared",\r
+ "HUB COMMERCIAL POWER FAIL - Fixed\_FWD",\r
\r
-"HUB COMMERCIAL POWER FAIL - Fixed\_FWD",\r
+ "HUB COMMERCIAL POWER FAIL - No Generator\_FWD",\r
\r
-"HUB COMMERCIAL POWER FAIL - No Generator\_FWD",\r
+ "HUB COMMERCIAL POWER FAIL - Portable\_FWD",\r
\r
-"HUB COMMERCIAL POWER FAIL - Portable\_FWD",\r
+ "HUB COMMERCIAL POWER FAIL - Shared\_FWD",\r
\r
-"HUB COMMERCIAL POWER FAIL - Shared\_FWD",\r
+ };\r
\r
-};\r
-\r
-}\r
+ }\r
\r
\r
\r
-** Filter Builder**\r
+**Filter Builder**\r
\r
MR server-side filtering allows a consumer to filter the stream of\r
-messages returned from the GET call. The following link provide details\r
-of building some of the filter to illustrate Filter Builder. It is not\r
-meant to cover and provide examples of every filter\r
+ messages returned from the GET call. The following link provide details\r
+ of building some of the filter to illustrate Filter Builder. It is not\r
+ meant to cover and provide examples of every filter\r
\r
\r
\r
Code Review / dmaap / messagerouter / messageservice.git / blobdiff