Add gauge support for the Counter Consumer

- Add Micrometer GAUGE Meter metrics. Add new counter.meter-type property to select between counter and gauge meter types.
 - Add tests and stock-exchange demo application.
 - Update counter-consumer README.

 Resolves: https://github.com/spring-cloud/stream-applications/issues/69

applications/processor/semantic-segmentation-processor/README.adoc

@@ -34,7 +34,7 @@ $$semantic.segmentation.color-map-uri$$:: $$Every pre-trained model is based on
 $$semantic.segmentation.debug-output$$:: $$save output image inn the local debugOutputPath path.$$ *($$Boolean$$, default: `$$false$$`)*
 $$semantic.segmentation.debug-output-path$$:: $$<documentation missing>$$ *($$String$$, default: `$$semantic-segmentation-result.png$$`)*
 $$semantic.segmentation.mask-transparency$$:: $$The alpha color of the computed segmentation mask image.$$ *($$Float$$, default: `$$0.45$$`)*
-$$semantic.segmentation.model$$:: $$pre-trained tensorflow semantic segmentation model.$$ *($$String$$, default: `$$http://download.tensorflow.org/models/deeplabv3_mnv2_cityscapes_train_2018_02_05.tar.gz#frozen_inference_graph.pb$$`)*
+$$semantic.segmentation.model$$:: $$pre-trained tensorflow semantic segmentation model.$$ *($$String$$, default: `$$https://download.tensorflow.org/models/deeplabv3_mnv2_cityscapes_train_2018_02_05.tar.gz#frozen_inference_graph.pb$$`)*
 $$semantic.segmentation.output-type$$:: $$Specifies the output image type. You can return either the input image with the computed mask overlay, or the mask alone.$$ *($$OutputType$$, default: `$$<none>$$`, possible values: `blended`,`mask`)*
 //end::configuration-properties[]
 

functions/consumer/counter-consumer/README.adoc

@@ -1,23 +1,170 @@
 # Counter Consumer
 
-A consumer that allows to compute multiple consumers from an incoming message.
-It uses micrometer internally and can use various popular TSDB technologies to persist the counter values.
+The `counter-consumer` is a Java https://docs.oracle.com/javase/8/docs/api/java/util/function/Consumer.html[Consumer<Message<?>>] that computes analytics from the input data messages and publishes them as metrics to various monitoring systems.
+It leverages the https://micrometer.io[micrometer library] for providing a uniform programming experience across the most popular https://micrometer.io/docs[monitoring systems] and uses https://docs.spring.io/spring-integration/reference/html/spel.html#spel[Spring Expression Language (SpEL)] for defining how the metric names, values and tags are computed from the input data.
 
-It is worth to look at the https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/counter-sink/README.adoc[counter-sink] where this consumer is used to create a Spring Cloud Stream based counter sink.
+The counter-consumer can produce two metrics types:
+
+- https://micrometer.io/docs/concepts#_counters[Counter] - reports a single metric, a count, that increments by a fixed, positive amount. Counters can be used for computing the rates of how the data changes in time.
+- https://micrometer.io/docs/concepts#_gauges[Gauge] - reports the current value. Typical examples for gauges would be the size of a collection or map or number of threads in a running state.
+
+A https://micrometer.io/docs/concepts#_meters[Meter] (e.g Counter or Gauge) is uniquely identified by its `name` and `dimensions` (the term dimensions and tags is used interchangeably). Dimensions allow a particular named metric to be sliced to drill down and reason about the data.
+
+NOTE: As a metrics is uniquely identified by its `name` and `dimensions`, you can assign multiple tags (e.g. key/value pairs) to every metric, but you cannot randomly change those tags afterwards! Monitoring systems such as Prometheus will complain if a metric with the same name has different sets of tags.
 
 ## Beans for injection
 
-You can import `CounterConsumerConfiguration` in the application and then inject the following bean.
+Add the counter-consumer dependency to your POM:
 
-`Consumer<Message<?>> counterConsumer`
+[source,xml]
+----
+<dependency>
+  <groupId>org.springframework.cloud.fn</groupId>
+  <artifactId>counter-consumer</artifactId>
+  <version>1.0.0-SNAPSHOT</version>
+</dependency>
+----
 
-You can use `counterConsumer` as a qualifier when injecting.
+Import the https://github.com/spring-cloud/stream-applications/blob/master/functions/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerConfiguration.java[CounterConsumerConfiguration] in the application and inject the following consumer bean:
+
+[source,java]
+----
+ Consumer<Message<?>> counterConsumer
+----
+
+For every input https://docs.spring.io/spring-integration/reference/html/message.html[Message] the `counterConsumer` computes the defined metrics and eventually, with the help of the micrometer library, publishes them to the backend monitoring systems. The https://docs.spring.io/spring-integration/reference/html/message.html[Message] is a generic container for data. Each Message instance includes a payload and headers containing user-extensible properties as key-value pairs.  Any object can be provided as the payload.
+The https://docs.spring.io/spring-integration/reference/html/message.html#message-builder[MessageBuilder] helps to create a Message instance from any payload content and assign any key/value as a header:
+
+[source,java]
+----
+   Message<String> myMessage = MessageBuilder
+      .withPayload("My message text")
+      .setHeader("kind", "CUSTOM")
+      .setHeader("foo", "bar")
+      .build();
+----
+
+The `SpEL` expressions use the `headers` and `payload` keywords to access message’s headers and payload values. For example a counter metrics can have a value amount computed from the size of the input message payload add a `my_tag` tag, extracted from the `kind` header value:
+
+[source]
+----
+counter.amount-expression=payload.lenght()
+counter.tag.expression.my_tag=headers['kind']
+----
+
+Review the https://github.com/spring-cloud/stream-applications/blob/master/functions/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerProperties.java[CounterConsumerProperties]'s javadocs for further details how to use the SpEL properties.
+
+By default, Micrometer is packed with a SimpleMeterRegistry that holds the latest value of each meter in memory and doesn’t export the data anywhere.
+To enable support for another monitoring system you have to add the spring-boot-starter-actuator dependency and the micrometer dependency of the monitoring system of choice:
+
+[source,xml]
+----
+ <dependency>
+    <groupId>org.springframework.boot</groupId>
+    <artifactId>spring-boot-starter-actuator</artifactId>
+    <version>CHANGE TO LATEST VERSION</version>
+ </dependency>
+
+ <dependency>
+   <groupId>io.micrometer</groupId>
+   <artifactId>micrometer-registry-[MONITORING SYSTEM NAME]</artifactId>
+   <version>${micrometer.version}</version>
+ </dependency>
+----
+
+Follow the https://docs.spring.io/spring-boot/docs/2.3.1.RELEASE/reference/html/production-ready-features.html#production-ready-metrics-export[configuration instructions] for the selected monitoring system. All monitoring configuration properties start with a prefix: `management.metrics.export`.
 
 ## Configuration Options
 
-All configuration properties are prefixed with `counter`.
+All `counter-consumer` configuration properties use the `counter` prefix. For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerProperties.java[CounterConsumerProperties].
+
+All monitoring configuration properties start with a prefix `management.metrics.export`. For configuring a particular monitoring system follow the provided https://docs.spring.io/spring-boot/docs/2.3.1.RELEASE/reference/html/production-ready-features.html#production-ready-metrics-export[configuration instructions].
+
+#### Sample Configuration
+
+Following examples show how to configure counter and gauge metrics over a series of stock-exchange messages like this:
+
+[source,json]
+----
+{
+  "data": {
+    "symbol": "AAPL",
+    "exchange": "XNAS",
+    "open": 318.66,
+    "close": 316.85,
+    "volume": 25672211.0
+  }
+}
+----
+
+The following configuration will create a counter metrics called `stockrates` with two tags: `symbol` and `exchange` computed from the json fields:
+
+.Counter Metrcis Configuration - count stock transactions
+|===
+|Property |Description
+
+|counter.meter-type=counter
+|Counter meter type (default)
+
+|counter.name=stockrates
+|Metrics name
+
+|counter.tag.expression.symbol=#jsonPath(payload,'$.data.symbol')
+|Add tag `symbol` equal to the `date.symbol` field in the json messages.
+
+|counter.tag.expression.exchange=#jsonPath(payload,'$.data.exchange')
+|Add tag `exchange` equal to the `date.exchange` field in the json messages.
+
+|===
+
+Now you can use the `stockrates` metrics to measure the rates at which the stock transactions occur over a given time interval. Furthermore, you can aggregate those rates by the `symbol` and `exchange` tags.
+
+To measure the transaction volumes contained in the data.volume JSON fields, you can build a GAUGE metrics like this:
+
+.Gauge Metrcis Configuration - compute stock volumes
+|===
+|Property |Description
+
+|counter.meter-type=gauge
+|Gauge meter type
+
+|counter.name=stockvolumes
+|Metrics name
+
+|counter.tag.expression.symbol=#jsonPath(payload,'$.data.symbol')
+|Add tag `symbol` equal to the `date.symbol` field in the json messages.
+
+|counter.tag.expression.exchange=#jsonPath(payload,'$.data.exchange')
+|Add tag `exchange` equal to the `date.exchange` field in the json messages.
+
+|counter.tag.amount-expression=#jsonPath(payload,'$.data.volume')
+|Set the Gauge to the `data/volume`  field values.
+|===
+
+Then use the `stockvolumes` metrics to graph, in real-time, the transaction volumes changes over time. You can aggregate those volumes by the `symbol` and `exchange` tags.
+
+WARNING: Micrometer implements the Gauges for the purpose of data sampling! There is no information about what might have occurred between two consecutive samples. Any intermediate values set on a gauge are lost by the time the gauge value is reported to a metrics backend.
+
+To enable one or more https://micrometer.io/docs[supported monitoring systems] you need to add a configuration like this:
+
+.Wavefront Configuration.
+|===
+|Property |Description
+
+|management.metrics.export.wavefront.enabled=true
+|Enable or disable the monitoring system. (enabled by default).
+
+|management.metrics.export.wavefront.uri=YOUR_WAVEFRONT_SERVER_URI
+|UIR of your Wavefront server or Wavefront Proxy.
+
+|management.metrics.export.wavefront.api-token=YOUR_API_TOKEN
+|Wavefront access token.
+
+|management.metrics.export.wavefront.source=stock-exchange-demo
+|The `source` is used to distinct your metrics on the Wavefront server.
+
+|===
 
-For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerProperties.java[CounterConsumerProperties].
 
 ## Tests
 
@@ -25,4 +172,6 @@ See this link:src/test/java/org/springframework/cloud/fn/consumer/counter[test s
 
 ## Other usage
 
-See this https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/counter-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream application where it makes a Counter sink.
+* See the https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/counter-sink/README.adoc[Counter Sink README] where this consumer is used to create a Spring Cloud Stream application where it makes a Counter sink.
+
+* https://docs.google.com/document/d/1BHBjgMmg4a1ue2wr-dmPTfgaN0so4ufw2XkG541Ac9Q/edit?usp=sharing[Stock Exchange Sample].

functions/consumer/counter-consumer/pom.xml

@@ -43,6 +43,21 @@
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-actuator</artifactId>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>io.micrometer</groupId>
+			<artifactId>micrometer-registry-wavefront</artifactId>
+			<version>1.5.1</version>
+			<scope>test</scope>
+		</dependency>
+
+
 	</dependencies>
 
 </project>

functions/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerConfiguration.java

@@ -22,10 +22,14 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
+import io.micrometer.core.instrument.Meter;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.Tags;
@@ -51,9 +55,11 @@ import org.springframework.util.StringUtils;
  * @author Christian Tzolov
  */
 @Configuration
-@EnableConfigurationProperties({CounterConsumerProperties.class})
+@EnableConfigurationProperties({ CounterConsumerProperties.class })
 public class CounterConsumerConfiguration {
 
+	private final Map<Meter.Id, AtomicLong> gaugeValues = new ConcurrentHashMap<>();
+
 	@Bean
 	public Function<String, Expression> stringToSpelFunction(@Lazy EvaluationContext evaluationContext) {
 		return new StringToSpelConversionFunction(evaluationContext);
@@ -72,19 +78,19 @@ public class CounterConsumerConfiguration {
 
 	@Bean(name = "counterConsumer")
 	public Consumer<Message<?>> counterConsumer(CounterConsumerProperties properties, MeterRegistry[] meterRegistries,
-												@Qualifier(IntegrationContextUtils.INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME) EvaluationContext context) {
+			@Qualifier(IntegrationContextUtils.INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME) EvaluationContext context) {
 
 		return message -> {
 
-			String counterName = properties.getComputedNameExpression().getValue(context, message, CharSequence.class).toString();
+			String meterName = properties.getComputedNameExpression().getValue(context, message, CharSequence.class).toString();
 
-			// All fixed tags together are passed with every counter increment.
+			// All fixed tags together are passed with every meter update.
 			Tags fixedTags = this.toTags(properties.getTag().getFixed());
 
 			double amount = properties.getComputedAmountExpression().getValue(context, message, double.class);
 
 			Map<String, List<Tag>> allGroupedTags = new HashMap<>();
-			// Tag Expressions Counter
+			// Tag Expressions
 			if (properties.getTag().getExpression() != null) {
 
 				Map<String, List<Tag>> groupedTags = properties.getTag().getExpression().entrySet().stream()
@@ -93,11 +99,11 @@ public class CounterConsumerConfiguration {
 								toList(namedExpression.getValue().getValue(context, message)).stream()
 										.map(tagValue -> Tag.of(namedExpression.getKey(), tagValue))
 										.collect(Collectors.toList())).flatMap(List::stream)
-						.collect(Collectors.groupingBy(tag -> tag.getKey(), Collectors.toList()));
+						.collect(Collectors.groupingBy(Tag::getKey, Collectors.toList()));
 				allGroupedTags.putAll(groupedTags);
 			}
 
-			this.count(meterRegistries, counterName, fixedTags, allGroupedTags, amount);
+			this.recordMetrics(meterRegistries, meterName, fixedTags, allGroupedTags, amount, properties.getMeterType());
 		};
 	}
 
@@ -117,7 +123,7 @@ public class CounterConsumerConfiguration {
 
 	/**
 	 * Converts the input value into an list of values. If the value is not a collection/array type the result
-	 * is a single element list. For collection/array input value the result is the list of stringifie content of
+	 * is a single element list. For collection/array input value the result is the list of stringified content of
 	 * this collection.
 	 *
 	 * @param value input value can be array, collection or single value.
@@ -133,17 +139,18 @@ public class CounterConsumerConfiguration {
 					: Arrays.asList(ObjectUtils.toObjectArray(value));
 
 			return valueCollection.stream()
-					.filter(v -> v != null)
+					.filter(Objects::nonNull)
 					.map(Object::toString)
 					.filter(StringUtils::hasText)
 					.collect(Collectors.toList());
 		}
 		else {
-			return Arrays.asList(value.toString());
+			return Collections.singletonList(value.toString());
 		}
 	}
 
-	private void count(MeterRegistry[] meterRegistries, String counterName, Tags fixedTags, Map<String, List<Tag>> groupedTags, double amount) {
+	private void recordMetrics(MeterRegistry[] meterRegistries, String meterName, Tags fixedTags, Map<String,
+			List<Tag>> groupedTags, double amount, CounterConsumerProperties.MeterType meterType) {
 		if (!CollectionUtils.isEmpty(groupedTags)) {
 			groupedTags.values().stream().map(List::size).max(Integer::compareTo).ifPresent(
 					max -> {
@@ -155,22 +162,48 @@ public class CounterConsumerConfiguration {
 										currentTags.and(Tags.of(e.getKey(), ""));
 							}
 
-							// Increment the counterName increment for every configured MaterRegistry.
-							for (MeterRegistry meterRegistry : meterRegistries) {
-								meterRegistry.counter(counterName, currentTags).increment(amount);
-							}
+							// Update the meterName for every configured MaterRegistry.
+							record(meterRegistries, meterName, currentTags, amount, meterType);
 						}
 					}
 			);
 		}
 		else {
-			// Increment the counterName increment for every configured MaterRegistry.
-			for (MeterRegistry meterRegistry : meterRegistries) {
-				meterRegistry.counter(counterName, fixedTags).increment(amount);
+			// Update the meterName for every configured MaterRegistry.
+			record(meterRegistries, meterName, fixedTags, amount, meterType);
+		}
+	}
+
+	private void record(MeterRegistry[] meterRegistries, String meterName,
+			Iterable<Tag> tags, double meterAmount, CounterConsumerProperties.MeterType meterType) {
+
+		for (MeterRegistry meterRegistry : meterRegistries) {
+			if (meterType == CounterConsumerProperties.MeterType.gauge) {
+				Meter.Id gaugeId = new Meter.Id(meterName, Tags.of(tags), null, null, Meter.Type.GAUGE);
+				if (!this.gaugeValues.containsKey(gaugeId)) {
+					this.gaugeValues.put(gaugeId, new AtomicLong((long) meterAmount));
+				}
+				else {
+					this.gaugeValues.get(gaugeId).set((long) meterAmount);
+				}
+				if (!isMeterRegistryContainsGauge(meterRegistry, gaugeId)) {
+					meterRegistry.gauge(meterName, tags, this.gaugeValues.get(gaugeId), AtomicLong::doubleValue);
+				}
+			}
+			else if (meterType == CounterConsumerProperties.MeterType.counter) {
+				meterRegistry.counter(meterName, tags).increment(meterAmount);
+			}
+			else {
+				throw new RuntimeException("Unknown meter type:" + meterType);
 			}
 		}
 	}
 
+	private boolean isMeterRegistryContainsGauge(MeterRegistry meterRegistry, Meter.Id gaugeId) {
+		return meterRegistry.find(gaugeId.getName()).gauges().stream()
+				.anyMatch(gauge -> gauge.getId().equals(gaugeId));
+	}
+
 	@Bean
 	@ConditionalOnMissingBean
 	public SimpleMeterRegistry simpleMeterRegistry() {

functions/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerProperties.java

@@ -33,6 +33,21 @@ import org.springframework.validation.annotation.Validated;
 @Validated
 public class CounterConsumerProperties {
 
+	enum MeterType {
+		/** Uses the Micrometer Counter meter type. It accumulates intermediate counts toward the point where
+		 * the data is sent to the metrics backend.*/
+		counter,
+		/** Uses the Micrometer Gauge meter type. Gauges sample the input time series. Any intermediate values set on
+		 * a gauge are lost by the time the gauge value is reported to a metrics backend.
+		 * TIP: Never gauge something you can count with a Counter!*/
+		gauge
+	}
+
+	/**
+	 * Micrometer meter type used to report the metrics to the backend.
+	 */
+	private MeterType meterType = MeterType.counter;
+
 	/**
 	 * The default name of the increment.
 	 */
@@ -66,12 +81,20 @@ public class CounterConsumerProperties {
 	/**
 	 * Fixed and computed tags to be assignee with the counter increment measurement.
 	 */
-	private MetricsTag tag = new MetricsTag();
+	private final MetricsTag tag = new MetricsTag();
 
 	public MetricsTag getTag() {
 		return tag;
 	}
 
+	public MeterType getMeterType() {
+		return meterType;
+	}
+
+	public void setMeterType(MeterType meterType) {
+		this.meterType = meterType;
+	}
+
 	public String getName() {
 		if (name == null && nameExpression == null) {
 			return defaultName;

functions/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerParentTest.java

@@ -27,7 +27,8 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.annotation.DirtiesContext;
 
-@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
+		properties = { "management.metrics.export.wavefront.enabled=false" })
 @DirtiesContext
 public class CounterConsumerParentTest {
 

functions/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/GaugeWithAmountTest.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.consumer.counter;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.context.TestPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Christian Tzolov
+ */
+@TestPropertySource(properties = {
+		"counter.meter-type=gauge",
+		"counter.name=myGauge",
+		"counter.tag.expression.foo='bar'",
+		"counter.amount-expression=payload.length()"
+})
+class GaugeWithAmountTest extends CounterConsumerParentTest {
+
+	@Test
+	void testCounterSink() {
+		String messageSmall = "hello";
+		counterConsumer.accept(new GenericMessage(messageSmall));
+		assertThat(meterRegistry.find("myGauge").gauge().value()).isEqualTo(size(messageSmall));
+
+		assertThat(meterRegistry.find("myGauge").gauge().getId().getTags()).hasSize(1);
+		assertThat(meterRegistry.find("myGauge").gauge().getId().getTag("foo")).isEqualTo("bar");
+
+		String messageMiddle = "hello world";
+		counterConsumer.accept(new GenericMessage(messageMiddle));
+		assertThat(meterRegistry.find("myGauge").gauge().value()).isEqualTo(size(messageMiddle));
+
+		String messageLarge = "hello world, hello people!";
+		counterConsumer.accept(new GenericMessage(messageLarge));
+		assertThat(meterRegistry.find("myGauge").gauge().value()).isEqualTo(size(messageLarge));
+
+	}
+
+	private double size(String msg) {
+		return Long.valueOf(msg.length()).doubleValue();
+	}
+}

functions/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/StockExchangeAnalyticsTests.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.consumer.counter;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Tag;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.core.io.DefaultResourceLoader;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.util.StreamUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Christian Tzolov
+ */
+
+@TestPropertySource(properties = {
+		"counter.meter-type=counter",
+		"counter.name=stocks",
+		"counter.tag.expression.symbol=#jsonPath(payload,'$.data.symbol')",
+		"counter.tag.expression.exchange=#jsonPath(payload,'$.data.exchange')"
+})
+public class StockExchangeAnalyticsTests extends CounterConsumerParentTest {
+
+	@Test
+	public void testCounter() throws IOException {
+		byte[] messageAppl = StreamUtils.copyToByteArray(
+				new DefaultResourceLoader().getResource("classpath:/data/stock_appl.json").getInputStream());
+
+		counterConsumer.accept(MessageBuilder.withPayload(messageAppl).build());
+		counterConsumer.accept(MessageBuilder.withPayload(messageAppl).build());
+		counterConsumer.accept(MessageBuilder.withPayload(messageAppl).build());
+
+		byte[] messageVmw = StreamUtils.copyToByteArray(
+				new DefaultResourceLoader().getResource("classpath:/data/stock_vmw.json").getInputStream());
+
+		counterConsumer.accept(MessageBuilder.withPayload(messageVmw).build());
+		counterConsumer.accept(MessageBuilder.withPayload(messageVmw).build());
+
+		Collection<Counter> counters = meterRegistry.find("stocks").counters();
+
+		assertThat(counters).hasSize(2);
+
+		Iterator<Counter> itr = counters.iterator();
+
+		Counter applCounter = itr.next();
+		assertThat(applCounter.count()).isEqualTo(3);
+		assertThat(applCounter.getId().getTags()).contains(Tag.of("symbol", "AAPL"), Tag.of("exchange", "XNAS"));
+
+		Counter vmwCounter = itr.next();
+		assertThat(vmwCounter.count()).isEqualTo(2);
+		assertThat(vmwCounter.getId().getTags()).contains(Tag.of("symbol", "VMW"), Tag.of("exchange", "NYSE"));
+
+	}
+}

functions/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/demo/StockExchangeAnalyticsExample.java

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.consumer.demo;
+
+import java.util.Random;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import io.micrometer.core.instrument.MeterRegistry;
+
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.fn.consumer.counter.CounterConsumerConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+
+/**
+ * Sample Spring Boot Application that uses the counterConsumer to compute running stats from
+ * stock exchange messages.
+ *
+ * Counter configuration:
+ * <code>
+ *    --counter.meter-type=counter
+ *    --counter.name=stocks
+ *    --counter.tag.expression.symbol=#jsonPath(payload,'$.data.symbol')
+ *    --counter.tag.expression.exchange=#jsonPath(payload,'$.data.exchange')
+ * </code>
+ *
+ * Gauge configuration:
+ * <code>
+ *    --counter.meter-type=gauge
+ *    --counter.name=stocks
+ *    --counter.tag.expression.symbol=#jsonPath(payload,'$.data.symbol')
+ *    --counter.tag.expression.exchange=#jsonPath(payload,'$.data.exchange')
+ *    --counter.amount-expression=#jsonPath(payload,'$.data.volume')
+ * </code>
+ *
+ * Sample Wavefront configuration:
+ * <code>
+ *    --management.metrics.export.wavefront.enabled=true
+ *    --management.metrics.export.wavefront.uri=YOUR_WAVEFRONT_SERVER_URI
+ *    --management.metrics.export.wavefront.api-token=YOUR_API_TOKEN
+ *    --management.metrics.export.wavefront.source=stock-exchange-demo
+ * </code>
+ *
+ * @author Christian Tzolov
+ */
+@Import(CounterConsumerConfiguration.class)
+@SpringBootApplication
+public class StockExchangeAnalyticsExample {
+
+	public static void main(String[] args) {
+		SpringApplication.run(StockExchangeAnalyticsExample.class, args);
+	}
+
+	@Bean
+	public CommandLineRunner commandLineRunner(Consumer<Message<?>> counterConsumer,
+			MeterRegistry meterRegistry, Supplier<String> stockMessageGenerator) {
+
+		// Run every second.
+		return args -> Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
+
+			String message = stockMessageGenerator.get();
+
+			// Submit new message using the stockMessageGenerator to generate random stock messages.
+			counterConsumer.accept(MessageBuilder.withPayload(message).build());
+
+			// Print current stock meters
+			System.out.println(meterRegistry.getMeters().stream()
+					.filter(meter -> meter.getId().getName().contains("stocks"))
+					.map(meter -> meter.getId().getType() + " | " + meter.getId() + " | " + meter.measure())
+					.collect(Collectors.joining("\n")) +
+					"\n=========================================================================");
+
+		}, 0, 1000, TimeUnit.MILLISECONDS);
+	}
+
+	@Bean
+	public Supplier<String> stockMessageGenerator() {
+		final Random random = new Random();
+		final String[][] STOCKS = new String[][] { { "NASDAQ", "GOOGL" }, { "AMS", "TOM2" }, { "NYSE", "CLDR" },
+				{ "NYSE", "VMW" }, { "NYSE", "IBM" }, { "NASDAQ", "MSFT" }, { "NASDAQ", "AAPL" } };
+
+		return () -> {
+			int stockIndex = random.nextInt(STOCKS.length);
+			return "{\n" +
+					"  \"data\": {\n" +
+					"      \"symbol\": \"" + STOCKS[stockIndex][1] + "\",\n" +
+					"      \"exchange\": \"" + STOCKS[stockIndex][0] + "\",\n" +
+					"      \"open\": " + (1 + 10 * random.nextDouble()) + ",\n" +
+					"      \"close\": " + (1 + 10 * random.nextDouble()) + ",\n" +
+					"      \"volume\": " + (1000 + 100000 * random.nextDouble()) + "\n" +
+					"  }\n" +
+					"}";
+		};
+	}
+}
+

functions/consumer/counter-consumer/src/test/resources/data/stock_appl.json

@@ -0,0 +1,12 @@
+{
+  "data": {
+      "date": "2020-05-21T00:00:00+0000",
+      "symbol": "AAPL",
+      "exchange": "XNAS",
+      "open": 318.66,
+      "high": 320.89,
+      "low": 315.87,
+      "close": 316.85,
+      "volume": 25672211.0
+  }
+}

functions/consumer/counter-consumer/src/test/resources/data/stock_vmw.json

@@ -0,0 +1,12 @@
+{
+  "data": {
+      "date": "2020-05-21T00:00:00+0000",
+      "symbol": "VMW",
+      "exchange": "NYSE",
+      "open": 160.16,
+      "high": 180.91,
+      "low": 165.87,
+      "close": 170.85,
+      "volume": 5672211.0
+  }
+}