• Spring Cloud Data Flow installed locally Follow the installation instructions to run Spring Cloud Data Flow on a local host.

• Running instance of Kafka

• Running instance of Apache Cassandra

• A database utility tool such as DBeaver to connect to the Cassandra instance. You might have to provide host, port, username and password depending on the Cassandra configuration you are using.

• Create a keyspace and a book table in Cassandra using:

CREATE KEYSPACE clouddata WITH REPLICATION = { 'class' : 'org.apache.cassandra.locator.SimpleStrategy', 'replication_factor': '1' } AND DURABLE_WRITES = true;
USE clouddata;
CREATE TABLE book  (
    id          uuid PRIMARY KEY,
    isbn        text,
    author      text,
    title       text
);

1. Register the out-of-the-box applications for the Kafka binder

   These samples assume that the Data Flow Server can access a remote Maven repository, repo.spring.io/libs-release by default. If your Data Flow server is running behind a firewall, or you are using a maven proxy preventing access to public repositories, you will need to install the sample apps in your internal Maven repository and configure the server accordingly. The sample applications are typically registered using Data Flow’s bulk import facility. For example, the Shell command dataflow:>app import --uri dataflow.spring.io/rabbitmq-maven-latest (The actual URI is release and binder specific so refer to the sample instructions for the actual URL). The bulk import URI references a plain text file containing entries for all of the publicly available Spring Cloud Stream and Task applications published to repo.spring.io. For example, source.http=maven://org.springframework.cloud.stream.app:http-source-rabbit:2.1.0.RELEASE registers the http source app at the corresponding Maven address, relative to the remote repository(ies) configured for the Data Flow server. The format is maven://<groupId>:<artifactId>:<version> You will need to download the required apps or build them and then install them in your Maven repository, using whatever group, artifact, and version you choose. If you do this, register individual apps using dataflow:>app register…​ using the maven:// resource URI format corresponding to your installed app.

   dataflow:>app import --uri https://dataflow.spring.io/kafka-maven-latest

2. Create the stream

   dataflow:>stream create cassandrastream --definition "http --server.port=8888 --spring.cloud.stream.bindings.output.contentType='application/json' | cassandra --ingestQuery='insert into book (id, isbn, title, author) values (uuid(), ?, ?, ?)' --keyspace=clouddata" --deploy
   
   Created and deployed new stream 'cassandrastream'

   If Cassandra isn’t running on default port on localhost or if you need username and password to connect, use one of the following options to specify the necessary connection parameters: --username='<USERNAME>' --password='<PASSWORD>' --port=<PORT> --contact-points=<LIST-OF-HOSTS>

3. Verify the stream is successfully deployed

4. Notice that cassandrastream-http and cassandrastream-cassandra Spring Cloud Stream applications are running as Spring Boot applications within the server as a collocated process.

   2015-12-15 15:52:31.576  INFO 18337 --- [nio-9393-exec-1] o.s.c.d.a.s.l.OutOfProcessModuleDeployer : deploying module org.springframework.cloud.stream.module:cassandra-sink:jar:exec:1.0.0.BUILD-SNAPSHOT instance 0
      Logs will be in /var/folders/c3/ctx7_rns6x30tq7rb76wzqwr0000gp/T/spring-cloud-data-flow-284240942697761420/cassandrastream.cassandra
   2015-12-15 15:52:31.583  INFO 18337 --- [nio-9393-exec-1] o.s.c.d.a.s.l.OutOfProcessModuleDeployer : deploying module org.springframework.cloud.stream.module:http-source:jar:exec:1.0.0.BUILD-SNAPSHOT instance 0
      Logs will be in /var/folders/c3/ctx7_rns6x30tq7rb76wzqwr0000gp/T/spring-cloud-data-flow-284240942697761420/cassandrastream.http

5. Post sample data pointing to the http endpoint: localhost:8888 (8888 is the server.port we specified for the http source in this case)

   dataflow:>http post --contentType 'application/json' --data '{"isbn": "1599869772", "title": "The Art of War", "author": "Sun Tzu"}' --target http://localhost:8888
   > POST (application/json;charset=UTF-8) http://localhost:8888 {"isbn": "1599869772", "title": "The Art of War", "author": "Sun Tzu"}
   > 202 ACCEPTED

6. Connect to the Cassandra instance and query the table clouddata.book to list the persisted records

   select * from clouddata.book;

7. You’re done!

• Cloud Foundry instance

• A rabbit service instance

• A Running instance of cassandra in Cloud Foundry or from another Cloud provider

• A database utility tool such as DBeaver to connect to the Cassandra instance. You might have to provide host, port, username and password depending on the Cassandra configuration you are using.

• Create a book table in your Cassandra keyspace using:

  CREATE TABLE book  (
      id          uuid PRIMARY KEY,
      isbn        text,
      author      text,
      title       text
  );

• Spring Cloud Data Flow installed on Cloud Foundry

1. Register the out-of-the-box applications for the Rabbit binder

   These samples assume that the Data Flow Server can access a remote Maven repository, repo.spring.io/libs-release by default. If your Data Flow server is running behind a firewall, or you are using a maven proxy preventing access to public repositories, you will need to install the sample apps in your internal Maven repository and configure the server accordingly. The sample applications are typically registered using Data Flow’s bulk import facility. For example, the Shell command dataflow:>app import --uri dataflow.spring.io/rabbitmq-maven-latest (The actual URI is release and binder specific so refer to the sample instructions for the actual URL). The bulk import URI references a plain text file containing entries for all of the publicly available Spring Cloud Stream and Task applications published to repo.spring.io. For example, source.http=maven://org.springframework.cloud.stream.app:http-source-rabbit:2.1.0.RELEASE registers the http source app at the corresponding Maven address, relative to the remote repository(ies) configured for the Data Flow server. The format is maven://<groupId>:<artifactId>:<version> You will need to download the required apps or build them and then install them in your Maven repository, using whatever group, artifact, and version you choose. If you do this, register individual apps using dataflow:>app register…​ using the maven:// resource URI format corresponding to your installed app.

   dataflow:>app import --uri https://dataflow.spring.io/rabbitmq-maven-latest

2. Create the stream

   dataflow:>stream create cassandrastream --definition "http --spring.cloud.stream.bindings.output.contentType='application/json' | cassandra --ingestQuery='insert into book (id, isbn, title, author) values (uuid(), ?, ?, ?)' --username='<USERNAME>' --password='<PASSWORD>' --port=<PORT> --contact-points=<HOST> --keyspace='<KEYSPACE>'" --deploy
   
   Created and deployed new stream 'cassandrastream'

$ cf apps
Getting apps in org [your-org] / space [your-space] as user...
OK

name                        requested state   instances   memory   disk   urls
cassandrastream-cassandra   started           1/1         1G       1G     cassandrastream-cassandra.app.io
cassandrastream-http        started           1/1         1G       1G     cassandrastream-http.app.io
dataflow-server             started           1/1         1G       1G     dataflow-server.app.io

http post --contentType 'application/json' --data '{"isbn": "1599869772", "title": "The Art of War", "author": "Sun Tzu"}' --target http://<YOUR-cassandrastream-http-APP-URL>
> POST (application/json;charset=UTF-8) https://cassandrastream-http.app.io {"isbn": "1599869772", "title": "The Art of War", "author": "Sun Tzu"}
> 202 ACCEPTED

$ cf scale cassandrastream-http -i 3
Scaling app cassandrastream-http in org user-dataflow / space development as user...
OK

$ cf apps
Getting apps in org user-dataflow / space development as user...
OK

name                        requested state   instances   memory   disk   urls
cassandrastream-cassandra   started           1/1         1G       1G     cassandrastream-cassandra.app.io
cassandrastream-http        started           3/3         1G       1G     cassandrastream-http.app.io
dataflow-server             started           1/1         1G       1G     dataflow-server.app.io

• Spring Cloud Data Flow installed locally Follow the installation instructions to run Spring Cloud Data Flow on a local host.

• Running instance of Kafka

• Running instance of MySQL

• A database utility tool such as DBeaver or DbVisualizer

• Create the test database with a names table (in MySQL) using:

  CREATE DATABASE test;
  USE test;
  CREATE TABLE names
  (
  	name varchar(255)
  );

1. Register the out-of-the-box applications for the Kafka binder

   These samples assume that the Data Flow Server can access a remote Maven repository, repo.spring.io/libs-release by default. If your Data Flow server is running behind a firewall, or you are using a maven proxy preventing access to public repositories, you will need to install the sample apps in your internal Maven repository and configure the server accordingly. The sample applications are typically registered using Data Flow’s bulk import facility. For example, the Shell command dataflow:>app import --uri dataflow.spring.io/rabbitmq-maven-latest (The actual URI is release and binder specific so refer to the sample instructions for the actual URL). The bulk import URI references a plain text file containing entries for all of the publicly available Spring Cloud Stream and Task applications published to repo.spring.io. For example, source.http=maven://org.springframework.cloud.stream.app:http-source-rabbit:2.1.0.RELEASE registers the http source app at the corresponding Maven address, relative to the remote repository(ies) configured for the Data Flow server. The format is maven://<groupId>:<artifactId>:<version> You will need to download the required apps or build them and then install them in your Maven repository, using whatever group, artifact, and version you choose. If you do this, register individual apps using dataflow:>app register…​ using the maven:// resource URI format corresponding to your installed app.

   dataflow:>app import --uri https://dataflow.spring.io/kafka-maven-latest

2. Create the stream

   dataflow:>stream create --name mysqlstream --definition "http --server.port=8787 | jdbc --tableName=names --columns=name --spring.datasource.driver-class-name=org.mariadb.jdbc.Driver --spring.datasource.url='jdbc:mysql://localhost:3306/test'" --deploy
   
   Created and deployed new stream 'mysqlstream'

   If MySQL isn’t running on default port on localhost or if you need username and password to connect, use one of the following options to specify the necessary connection parameters: --spring.datasource.url='jdbc:mysql://<HOST>:<PORT>/<NAME>' --spring.datasource.username=<USERNAME> --spring.datasource.password=<PASSWORD>

3. Verify the stream is successfully deployed

4. Notice that mysqlstream-http and mysqlstream-jdbc Spring Cloud Stream applications are running as Spring Boot applications within the Local server as collocated processes.

   2016-05-03 09:29:55.918  INFO 65162 --- [nio-9393-exec-3] o.s.c.d.spi.local.LocalAppDeployer       : deploying app mysqlstream.jdbc instance 0
      Logs will be in /var/folders/c3/ctx7_rns6x30tq7rb76wzqwr0000gp/T/spring-cloud-dataflow-6850863945840320040/mysqlstream1-1462292995903/mysqlstream.jdbc
   2016-05-03 09:29:55.939  INFO 65162 --- [nio-9393-exec-3] o.s.c.d.spi.local.LocalAppDeployer       : deploying app mysqlstream.http instance 0
      Logs will be in /var/folders/c3/ctx7_rns6x30tq7rb76wzqwr0000gp/T/spring-cloud-dataflow-6850863945840320040/mysqlstream-1462292995934/mysqlstream.http

5. Post sample data pointing to the http endpoint: localhost:8787 [8787 is the server.port we specified for the http source in this case]

   dataflow:>http post --contentType 'application/json' --target http://localhost:8787 --data "{\"name\": \"Foo\"}"
   > POST (application/json;charset=UTF-8) http://localhost:8787 {"name": "Spring Boot"}
   > 202 ACCEPTED

6. Connect to the MySQL instance and query the table test.names to list the new rows:

   select * from test.names;

7. You’re done!

• Cloud Foundry instance

• Running instance of rabbit in Cloud Foundry

• Running instance of mysql in Cloud Foundry

• A database utility tool such as DBeaver or DbVisualizer

• Create the names table (in MySQL) using:

  CREATE TABLE names
  (
  	name varchar(255)
  );

• Spring Cloud Data Flow installed on Cloud Foundry

1. Register the out-of-the-box applications for the Rabbit binder

   These samples assume that the Data Flow Server can access a remote Maven repository, repo.spring.io/libs-release by default. If your Data Flow server is running behind a firewall, or you are using a maven proxy preventing access to public repositories, you will need to install the sample apps in your internal Maven repository and configure the server accordingly. The sample applications are typically registered using Data Flow’s bulk import facility. For example, the Shell command dataflow:>app import --uri dataflow.spring.io/rabbitmq-maven-latest (The actual URI is release and binder specific so refer to the sample instructions for the actual URL). The bulk import URI references a plain text file containing entries for all of the publicly available Spring Cloud Stream and Task applications published to repo.spring.io. For example, source.http=maven://org.springframework.cloud.stream.app:http-source-rabbit:2.1.0.RELEASE registers the http source app at the corresponding Maven address, relative to the remote repository(ies) configured for the Data Flow server. The format is maven://<groupId>:<artifactId>:<version> You will need to download the required apps or build them and then install them in your Maven repository, using whatever group, artifact, and version you choose. If you do this, register individual apps using dataflow:>app register…​ using the maven:// resource URI format corresponding to your installed app.

   dataflow:>app import --uri https://dataflow.spring.io/rabbitmq-maven-latest

2. Create the stream

   dataflow:>stream create --name mysqlstream --definition "http | jdbc --tableName=names --columns=name"
   Created new stream 'mysqlstream'
   
   dataflow:>stream deploy --name mysqlstream --properties "deployer.jdbc.cloudfoundry.services=mysql"
   Deployed stream 'mysqlstream'

   By supplying the deployer.jdbc.cloudfoundry.services=mysql property, we are deploying the stream with jdbc-sink to automatically bind to mysql service and only this application in the stream gets the service binding. This also eliminates the requirement to supply datasource credentials in stream definition.

3. Verify the stream is successfully deployed

4. Notice that mysqlstream-http and mysqlstream-jdbc Spring Cloud Stream applications are running as cloud-native (microservice) applications in Cloud Foundry

   $ cf apps
   Getting apps in org user-dataflow / space development as user...
   OK
   
   name                        requested state   instances   memory   disk   urls
   mysqlstream-http            started           1/1         1G       1G     mysqlstream-http.app.io
   mysqlstream-jdbc            started           1/1         1G       1G     mysqlstream-jdbc.app.io
   dataflow-server             started           1/1         1G       1G     dataflow-server.app.io

5. Lookup the url for mysqlstream-http application from the list above. Post sample data pointing to the http endpoint: <YOUR-mysqlstream-http-APP-URL>

   http post --contentType 'application/json' --data "{\"name\": \"Bar"} --target https://mysqlstream-http.app.io "
   > POST (application/json;charset=UTF-8) https://mysqlstream-http.app.io {"name": "Bar"}
   > 202 ACCEPTED

6. Connect to the MySQL instance and query the table names to list the new rows:

7. Now, let’s take advantage of Pivotal Cloud Foundry’s platform capability. Let’s scale the mysqlstream-http application from 1 to 3 instances

   $ cf scale mysqlstream-http -i 3
   Scaling app mysqlstream-http in org user-dataflow / space development as user...
   OK

8. Verify App instances (3/3) running successfully

   $ cf apps
   Getting apps in org user-dataflow / space development as user...
   OK
   
   name                        requested state   instances   memory   disk   urls
   mysqlstream-http            started           3/3         1G       1G     mysqlstream-http.app.io
   mysqlstream-jdbc            started           1/1         1G       1G     mysqlstream-jdbc.app.io
   dataflow-server             started           1/1         1G       1G     dataflow-server.app.io

9. You’re done!

• Spring Cloud Data Flow installed locally Follow the installation instructions to run Spring Cloud Data Flow on a local host.

• A running instance of Rabbit MQ

1. Use gfsh to start a locator and server

   gfsh>start locator --name=locator1
   gfsh>start server --name=server1

2. Create a region called Stocks

   gfsh>create region --name Stocks --type=REPLICATE

   Use the Shell to create the sample stream

3. Register the out-of-the-box applications for the Rabbit binder

   These samples assume that the Data Flow Server can access a remote Maven repository, repo.spring.io/libs-release by default. If your Data Flow server is running behind a firewall, or you are using a maven proxy preventing access to public repositories, you will need to install the sample apps in your internal Maven repository and configure the server accordingly. The sample applications are typically registered using Data Flow’s bulk import facility. For example, the Shell command dataflow:>app import --uri dataflow.spring.io/rabbitmq-maven-latest (The actual URI is release and binder specific so refer to the sample instructions for the actual URL). The bulk import URI references a plain text file containing entries for all of the publicly available Spring Cloud Stream and Task applications published to repo.spring.io. For example, source.http=maven://org.springframework.cloud.stream.app:http-source-rabbit:2.1.0.RELEASE registers the http source app at the corresponding Maven address, relative to the remote repository(ies) configured for the Data Flow server. The format is maven://<groupId>:<artifactId>:<version> You will need to download the required apps or build them and then install them in your Maven repository, using whatever group, artifact, and version you choose. If you do this, register individual apps using dataflow:>app register…​ using the maven:// resource URI format corresponding to your installed app.

   dataflow:>app import --uri https://dataflow.spring.io/rabbitmq-maven-latest

4. Create the stream

   This example creates an http endpoint to which we will post stock prices as a JSON document containing symbol and price fields. The property --json=true to enable Geode’s JSON support and configures the sink to convert JSON String payloads to PdxInstance, the recommended way to store JSON documents in Geode. The keyExpression property is a SpEL expression used to extract the symbol value the PdxInstance to use as an entry key.

   PDX serialization is very efficient and supports OQL queries without requiring a custom domain class. Use of custom domain types requires these classes to be in the class path of both the stream apps and the cache server. For this reason, the use of custom payload types is generally discouraged.

   dataflow:>stream create --name stocks --definition "http --port=9090 | gemfire --json=true --regionName=Stocks --keyExpression=payload.getField('symbol')" --deploy
   Created and deployed new stream 'stocks'

   If the Geode locator isn’t running on default port on localhost, add the options --connect-type=locator --host-addresses=<host>:<port>. If there are multiple locators, you can provide a comma separated list of locator addresses. This is not necessary for the sample but is typical for production environments to enable fail-over.

5. Verify the stream is successfully deployed

6. Post sample data pointing to the http endpoint: localhost:9090 (9090 is the port we specified for the http source)

   dataflow:>http post --target http://localhost:9090 --contentType application/json --data '{"symbol":"VMW","price":117.06}'
   > POST (application/json) http://localhost:9090 {"symbol":"VMW","price":117.06}
   > 202 ACCEPTED

7. Using gfsh, connect to the locator if not already connected, and verify the cache entry was created.

   gfsh>get --key='VMW' --region=/Stocks
   Result      : true
   Key Class   : java.lang.String
   Key         : VMW
   Value Class : org.apache.geode.pdx.internal.PdxInstanceImpl
   
   symbol | price
   ------ | ------
   VMW    | 117.06

8. You’re done!

• A Cloud Foundry instance

• Running instance of a rabbit service in Cloud Foundry

• Running instance of the Pivotal Cloud Cache for PCF (PCC) service cloudcache in Cloud Foundry.

• Spring Cloud Data Flow installed on Cloud Foundry Follow the installation instructions to run Spring Cloud Data Flow on Cloud Foundry.

1. Register the out-of-the-box applications for the Rabbit binder

   These samples assume that the Data Flow Server can access a remote Maven repository, repo.spring.io/libs-release by default. If your Data Flow server is running behind a firewall, or you are using a maven proxy preventing access to public repositories, you will need to install the sample apps in your internal Maven repository and configure the server accordingly. The sample applications are typically registered using Data Flow’s bulk import facility. For example, the Shell command dataflow:>app import --uri dataflow.spring.io/rabbitmq-maven-latest (The actual URI is release and binder specific so refer to the sample instructions for the actual URL). The bulk import URI references a plain text file containing entries for all of the publicly available Spring Cloud Stream and Task applications published to repo.spring.io. For example, source.http=maven://org.springframework.cloud.stream.app:http-source-rabbit:2.1.0.RELEASE registers the http source app at the corresponding Maven address, relative to the remote repository(ies) configured for the Data Flow server. The format is maven://<groupId>:<artifactId>:<version> You will need to download the required apps or build them and then install them in your Maven repository, using whatever group, artifact, and version you choose. If you do this, register individual apps using dataflow:>app register…​ using the maven:// resource URI format corresponding to your installed app.

   dataflow:>app import --uri https://dataflow.spring.io/rabbitmq-maven-latest

2. Get the PCC connection information

   $ cf service-key cloudcache my-service-key
   Getting key my-service-key for service instance cloudcache as <user>...
   
   {
    "locators": [
     "10.0.16.9[55221]",
     "10.0.16.11[55221]",
     "10.0.16.10[55221]"
    ],
    "urls": {
     "gfsh": "http://...",
     "pulse": "https://.../pulse"
    },
    "users": [
     {
      "password": <password>,
      "username": "cluster_operator"
     },
     {
      "password": <password>,
      "username": "developer"
     }
    ]
   }

3. Using gfsh, connect to the PCC instance as cluster_operator using the service key values and create the Stocks region.

   gfsh>connect --use-http --url=<gfsh-url> --user=cluster_operator --password=<cluster_operator_password>
   gfsh>create region --name Stocks --type=REPLICATE

4. Create the stream, connecting to the PCC instance as developer

   This example creates an http endpoint to which we will post stock prices as a JSON document containing symbol and price fields. The property --json=true to enable Geode’s JSON support and configures the sink to convert JSON String payloads to PdxInstance, the recommended way to store JSON documents in Geode. The keyExpression property is a SpEL expression used to extract the symbol value the PdxInstance to use as an entry key.

   PDX serialization is very efficient and supports OQL queries without requiring a custom domain class. Use of custom domain types requires these classes to be in the class path of both the stream apps and the cache server. For this reason, the use of custom payload types is generally discouraged.

   dataflow:>stream create --name stocks --definition "http --security.basic.enabled=false | gemfire --username=developer --password=<developer-password> --connect-type=locator --host-addresses=10.0.16.9:55221 --json=true --regionName=Stocks --keyExpression=payload.getField('symbol')" --deploy

5. Verify the stream is successfully deployed

6. Post sample data pointing to the http endpoint

   Get the url of the http source using cf apps

   dataflow:>http post --target http://<http source url> --contentType application/json --data '{"symbol":"VMW","price":117.06}'
   > POST (application/json) http://... {"symbol":"VMW","price":117.06}
   > 202 ACCEPTED

7. Using gfsh, connect to the PCC instance as cluster_operator using the service key values.

   gfsh>connect --use-http --url=<gfsh-url> --user=cluster_operator --password=<cluster_operator_password>
   gfsh>get --key='VMW' --region=/Stocks
   Result      : true
   Key Class   : java.lang.String
   Key         : VMW
   Value Class : org.apache.geode.pdx.internal.PdxInstanceImpl
   
   symbol | price
   ------ | ------
   VMW    | 117.06

8. You’re done!

• Spring Cloud Data Flow installed locally Follow the installation instructions to run Spring Cloud Data Flow on a local host.

• A running instance of Rabbit MQ

1. Use gfsh to start a locator and server

   gfsh>start locator --name=locator1
   gfsh>start server --name=server1

2. Create a region called Orders

   gfsh>create region --name Orders --type=REPLICATE

   Use the Shell to create the sample stream

3. Register the out-of-the-box applications for the Rabbit binder

   These samples assume that the Data Flow Server can access a remote Maven repository, repo.spring.io/libs-release by default. If your Data Flow server is running behind a firewall, or you are using a maven proxy preventing access to public repositories, you will need to install the sample apps in your internal Maven repository and configure the server accordingly. The sample applications are typically registered using Data Flow’s bulk import facility. For example, the Shell command dataflow:>app import --uri dataflow.spring.io/rabbitmq-maven-latest (The actual URI is release and binder specific so refer to the sample instructions for the actual URL). The bulk import URI references a plain text file containing entries for all of the publicly available Spring Cloud Stream and Task applications published to repo.spring.io. For example, source.http=maven://org.springframework.cloud.stream.app:http-source-rabbit:2.1.0.RELEASE registers the http source app at the corresponding Maven address, relative to the remote repository(ies) configured for the Data Flow server. The format is maven://<groupId>:<artifactId>:<version> You will need to download the required apps or build them and then install them in your Maven repository, using whatever group, artifact, and version you choose. If you do this, register individual apps using dataflow:>app register…​ using the maven:// resource URI format corresponding to your installed app.

   dataflow:>app import --uri https://dataflow.spring.io/rabbitmq-maven-latest

4. Create the stream

   This example creates an gemfire-cq source to which will publish events matching a query criteria on a region. In this case we will monitor the Orders region. For simplicity, we will avoid creating a data structure for the order. Each cache entry contains an integer value representing the quantity of the ordered item. This stream will fire a message whenever the value>999. By default, the source emits only the value. Here we will override that using the cq-event-expression property. This accepts a SpEL expression bound to a CQEvent. To reference the entire CQEvent instance, we use #this. In order to display the contents in the log, we will invoke toString() on the instance.

   dataflow:>stream create --name orders --definition " gemfire-cq --query='SELECT * from /Orders o where o > 999' --cq-event-expression=#this.toString() | log" --deploy

   If the Geode locator isn’t running on default port on localhost, add the options --connect-type=locator --host-addresses=<host>:<port>. If there are multiple locators, you can provide a comma separated list of locator addresses. This is not necessary for the sample but is typical for production environments to enable fail-over.

5. Verify the stream is successfully deployed

6. Monitor stdout for the log sink. When you deploy the stream, you will see log messages in the Data Flow server console like this

   2017-10-30 09:39:36.283  INFO 8167 --- [nio-9393-exec-5] o.s.c.d.spi.local.LocalAppDeployer       : Deploying app with deploymentId orders.log instance 0.
      Logs will be in /var/folders/hd/5yqz2v2d3sxd3n879f4sg4gr0000gn/T/spring-cloud-dataflow-5375107584795488581/orders-1509370775940/orders.log

   Copy the location of the log sink logs. This is a directory that ends in orders.log. The log files will be in stdout_0.log under this directory. You can monitor the output of the log sink using tail, or something similar:

   $tail -f /var/folders/hd/5yqz2v2d3sxd3n879f4sg4gr0000gn/T/spring-cloud-dataflow-5375107584795488581/orders-1509370775940/orders.log/stdout_0.log

7. Using gfsh, create and update some cache entries

   gfsh>put --region Orders  --value-class java.lang.Integer --key 01234 --value 1000
   gfsh>put --region Orders  --value-class java.lang.Integer --key 11234 --value 1005
   gfsh>put --region Orders  --value-class java.lang.Integer --key 21234 --value 100
   gfsh>put --region Orders  --value-class java.lang.Integer --key 31234 --value 999
   gfsh>put --region Orders  --value-class java.lang.Integer --key 21234 --value 1000

8. Observe the log output You should see messages like:

   2017-10-30 09:53:02.231  INFO 8563 --- [ire-cq.orders-1] log-sink                                 : CqEvent [CqName=GfCq1; base operation=CREATE; cq operation=CREATE; key=01234; value=1000]
   2017-10-30 09:53:19.732  INFO 8563 --- [ire-cq.orders-1] log-sink                                 : CqEvent [CqName=GfCq1; base operation=CREATE; cq operation=CREATE; key=11234; value=1005]
   2017-10-30 09:53:53.242  INFO 8563 --- [ire-cq.orders-1] log-sink                                 : CqEvent [CqName=GfCq1; base operation=UPDATE; cq operation=CREATE; key=21234; value=1000]

9. Another interesting demonstration combines gemfire-cq with the http-gemfire example.

dataflow:> stream create --name stocks --definition "http --port=9090 | gemfire-json-server --regionName=Stocks --keyExpression=payload.getField('symbol')" --deploy
dataflow:> stream create --name stock_watch --definition "gemfire-cq --query='Select * from /Stocks where symbol=''VMW''' | log" --deploy

1. You’re done!

• A Cloud Foundry instance

• Running instance of a rabbit service in Cloud Foundry

• Running instance of the Pivotal Cloud Cache for PCF (PCC) service cloudcache in Cloud Foundry.

• Spring Cloud Data Flow installed on Cloud Foundry

1. Register the out-of-the-box applications for the Rabbit binder

   These samples assume that the Data Flow Server can access a remote Maven repository, repo.spring.io/libs-release by default. If your Data Flow server is running behind a firewall, or you are using a maven proxy preventing access to public repositories, you will need to install the sample apps in your internal Maven repository and configure the server accordingly. The sample applications are typically registered using Data Flow’s bulk import facility. For example, the Shell command dataflow:>app import --uri dataflow.spring.io/rabbitmq-maven-latest (The actual URI is release and binder specific so refer to the sample instructions for the actual URL). The bulk import URI references a plain text file containing entries for all of the publicly available Spring Cloud Stream and Task applications published to repo.spring.io. For example, source.http=maven://org.springframework.cloud.stream.app:http-source-rabbit:2.1.0.RELEASE registers the http source app at the corresponding Maven address, relative to the remote repository(ies) configured for the Data Flow server. The format is maven://<groupId>:<artifactId>:<version> You will need to download the required apps or build them and then install them in your Maven repository, using whatever group, artifact, and version you choose. If you do this, register individual apps using dataflow:>app register…​ using the maven:// resource URI format corresponding to your installed app.

   dataflow:>app import --uri https://dataflow.spring.io/rabbitmq-maven-latest

2. Get the PCC connection information

   $ cf service-key cloudcache my-service-key
   Getting key my-service-key for service instance cloudcache as <user>...
   
   {
    "locators": [
     "10.0.16.9[55221]",
     "10.0.16.11[55221]",
     "10.0.16.10[55221]"
    ],
    "urls": {
     "gfsh": "http://...",
     "pulse": "https://.../pulse"
    },
    "users": [
     {
      "password": <password>,
      "username": "cluster_operator"
     },
     {
      "password": <password>,
      "username": "developer"
     }
    ]
   }

3. Using gfsh, connect to the PCC instance as cluster_operator using the service key values and create the Test region.

   gfsh>connect --use-http --url=<gfsh-url> --user=cluster_operator --password=<cluster_operator_password>
   gfsh>create region --name Orders --type=REPLICATE

4. Create the stream using the Data Flow Shell

   This example creates an gemfire-cq source to which will publish events matching a query criteria on a region. In this case we will monitor the Orders region. For simplicity, we will avoid creating a data structure for the order. Each cache entry contains an integer value representing the quantity of the ordered item. This stream will fire a message whenever the value>999. By default, the source emits only the value. Here we will override that using the cq-event-expression property. This accepts a SpEL expression bound to a CQEvent. To reference the entire CQEvent instance, we use #this. In order to display the contents in the log, we will invoke toString() on the instance.

   dataflow:>stream create --name orders --definition " gemfire-cq  --username=developer --password=<developer-password> --connect-type=locator --host-addresses=10.0.16.9:55221 --query='SELECT * from /Orders o where o > 999' --cq-event-expression=#this.toString()  | log" --deploy
   Created and deployed new stream 'events'

5. Verify the stream is successfully deployed

6. Monitor stdout for the log sink

   cf logs <log-sink-app-name>

7. Using gfsh, create and update some cache entries

   gfsh>connect --use-http --url=<gfsh-url> --user=cluster_operator --password=<cluster_operator_password>
   gfsh>put --region Orders  --value-class java.lang.Integer --key 01234 --value 1000
   gfsh>put --region Orders  --value-class java.lang.Integer --key 11234 --value 1005
   gfsh>put --region Orders  --value-class java.lang.Integer --key 21234 --value 100
   gfsh>put --region Orders  --value-class java.lang.Integer --key 31234 --value 999
   gfsh>put --region Orders  --value-class java.lang.Integer --key 21234 --value 1000

8. Observe the log output You should see messages like:

   2017-10-30 09:53:02.231  INFO 8563 --- [ire-cq.orders-1] log-sink                                 : CqEvent [CqName=GfCq1; base operation=CREATE; cq operation=CREATE; key=01234; value=1000]
   2017-10-30 09:53:19.732  INFO 8563 --- [ire-cq.orders-1] log-sink                                 : CqEvent [CqName=GfCq1; base operation=CREATE; cq operation=CREATE; key=11234; value=1005]
   2017-10-30 09:53:53.242  INFO 8563 --- [ire-cq.orders-1] log-sink                                 : CqEvent [CqName=GfCq1; base operation=UPDATE; cq operation=CREATE; key=21234; value=1000]

9. Another interesting demonstration combines gemfire-cq with the http-gemfire example.

   dataflow:>stream create --name stocks --definition "http --security.basic.enabled=false | gemfire --json=true --username=developer --password=<developer-password> --connect-type=locator --host-addresses=10.0.16.25:55221 --regionName=Stocks --keyExpression=payload.getField('symbol')" --deploy
   
   dataflow:>stream create --name stock_watch --definition "gemfire-cq  --username=developer --password=<developer-password> --connect-type=locator --host-addresses=10.0.16.25:55221 --query='SELECT * from /Stocks where symbol=''VMW''' --cq-event-expression=#this.toString()  | log" --deploy
   
   dataflow:>http post --target http://data-flow-server-dpvuo77-stocks-http.apps.scdf-gcp.springapps.io/ --contentType application/json --data '{"symbol":"VMW","price":117.06}'

10. You’re done!

• Spring Cloud Data Flow installed locally Follow the installation instructions to run Spring Cloud Data Flow on a local host.

• A running instance of Rabbit MQ

1. Use gfsh to start a locator and server

   gfsh>start locator --name=locator1
   gfsh>start server --name=server1

2. Create a region called Test

   gfsh>create region --name Test --type=REPLICATE

   Use the Shell to create the sample stream

3. Register the out-of-the-box applications for the Rabbit binder

   These samples assume that the Data Flow Server can access a remote Maven repository, repo.spring.io/libs-release by default. If your Data Flow server is running behind a firewall, or you are using a maven proxy preventing access to public repositories, you will need to install the sample apps in your internal Maven repository and configure the server accordingly. The sample applications are typically registered using Data Flow’s bulk import facility. For example, the Shell command dataflow:>app import --uri dataflow.spring.io/rabbitmq-maven-latest (The actual URI is release and binder specific so refer to the sample instructions for the actual URL). The bulk import URI references a plain text file containing entries for all of the publicly available Spring Cloud Stream and Task applications published to repo.spring.io. For example, source.http=maven://org.springframework.cloud.stream.app:http-source-rabbit:2.1.0.RELEASE registers the http source app at the corresponding Maven address, relative to the remote repository(ies) configured for the Data Flow server. The format is maven://<groupId>:<artifactId>:<version> You will need to download the required apps or build them and then install them in your Maven repository, using whatever group, artifact, and version you choose. If you do this, register individual apps using dataflow:>app register…​ using the maven:// resource URI format corresponding to your installed app.

   dataflow:>app import --uri https://dataflow.spring.io/rabbitmq-maven-latest

4. Create the stream

   This example creates an gemfire source to which will publish events on a region

   dataflow:>stream create --name events --definition " gemfire --regionName=Test | log" --deploy
   Created and deployed new stream 'events'

   If the Geode locator isn’t running on default port on localhost, add the options --connect-type=locator --host-addresses=<host>:<port>. If there are multiple locators, you can provide a comma separated list of locator addresses. This is not necessary for the sample but is typical for production environments to enable fail-over.

5. Verify the stream is successfully deployed

6. Monitor stdout for the log sink. When you deploy the stream, you will see log messages in the Data Flow server console like this

   2017-10-28 17:28:23.275  INFO 15603 --- [nio-9393-exec-2] o.s.c.d.spi.local.LocalAppDeployer       : Deploying app with deploymentId events.log instance 0.
      Logs will be in /var/folders/hd/5yqz2v2d3sxd3n879f4sg4gr0000gn/T/spring-cloud-dataflow-4093992067314402881/events-1509226103269/events.log
   2017-10-28 17:28:23.277  INFO 15603 --- [nio-9393-exec-2] o.s.c.d.s.c.StreamDeploymentController   : Downloading resource URI [maven://org.springframework.cloud.stream.app:gemfire-source-rabbit:1.2.0.RELEASE]
   2017-10-28 17:28:23.311  INFO 15603 --- [nio-9393-exec-2] o.s.c.d.s.c.StreamDeploymentController   : Deploying application named [gemfire] as part of stream named [events] with resource URI [maven://org.springframework.cloud.stream.app:gemfire-source-rabbit:1.2.0.RELEASE]
   2017-10-28 17:28:23.318  INFO 15603 --- [nio-9393-exec-2] o.s.c.d.spi.local.LocalAppDeployer       : Deploying app with deploymentId events.gemfire instance 0.
      Logs will be in /var/folders/hd/5yqz2v2d3sxd3n879f4sg4gr0000gn/T/spring-cloud-dataflow-4093992067314402881/events-1509226103311/events.gemfire

   Copy the location of the log sink logs. This is a directory that ends in events.log. The log files will be in stdout_0.log under this directory. You can monitor the output of the log sink using tail, or something similar:

   $tail -f /var/folders/hd/5yqz2v2d3sxd3n879f4sg4gr0000gn/T/spring-cloud-dataflow-4093992067314402881/events-1509226103269/events.log/stdout_0.log

7. Using gfsh, create and update some cache entries

   gfsh>put --region /Test --key 1  --value "value 1"
   gfsh>put --region /Test --key 2  --value "value 2"
   gfsh>put --region /Test --key 3  --value "value 3"
   gfsh>put --region /Test --key 1  --value "new value 1"

8. Observe the log output You should see messages like:

   2017-10-28 17:28:52.893  INFO 18986 --- [emfire.events-1] log sink                               : value 1"
   2017-10-28 17:28:52.893  INFO 18986 --- [emfire.events-1] log sink                               : value 2"
   2017-10-28 17:28:52.893  INFO 18986 --- [emfire.events-1] log sink                               : value 3"
   2017-10-28 17:28:52.893  INFO 18986 --- [emfire.events-1] log sink                               : new value 1"

   By default, the message payload contains the updated value. Depending on your application, you may need additional information. The data comes from EntryEvent. You can access any fields using the source’s cache-event-expression property. This takes a SpEL expression bound to the EntryEvent. Try something like --cache-event-expression='{key:'+key+',new_value:'+newValue+'}' (HINT: You will need to destroy the stream and recreate it to add this property, an exercise left to the reader). Now you should see log messages like:

   2017-10-28 17:28:52.893  INFO 18986 --- [emfire.events-1] log-sink                                 : {key:1,new_value:value 1}
   2017-10-28 17:41:24.466  INFO 18986 --- [emfire.events-1] log-sink                                 : {key:2,new_value:value 2}

9. You’re done!

• A Cloud Foundry instance

• Running instance of a rabbit service in Cloud Foundry

• Running instance of the Pivotal Cloud Cache for PCF (PCC) service cloudcache in Cloud Foundry.

• Spring Cloud Data Flow installed on Cloud Foundry

1. Register the out-of-the-box applications for the Rabbit binder

   These samples assume that the Data Flow Server can access a remote Maven repository, repo.spring.io/libs-release by default. If your Data Flow server is running behind a firewall, or you are using a maven proxy preventing access to public repositories, you will need to install the sample apps in your internal Maven repository and configure the server accordingly. The sample applications are typically registered using Data Flow’s bulk import facility. For example, the Shell command dataflow:>app import --uri dataflow.spring.io/rabbitmq-maven-latest (The actual URI is release and binder specific so refer to the sample instructions for the actual URL). The bulk import URI references a plain text file containing entries for all of the publicly available Spring Cloud Stream and Task applications published to repo.spring.io. For example, source.http=maven://org.springframework.cloud.stream.app:http-source-rabbit:2.1.0.RELEASE registers the http source app at the corresponding Maven address, relative to the remote repository(ies) configured for the Data Flow server. The format is maven://<groupId>:<artifactId>:<version> You will need to download the required apps or build them and then install them in your Maven repository, using whatever group, artifact, and version you choose. If you do this, register individual apps using dataflow:>app register…​ using the maven:// resource URI format corresponding to your installed app.

   dataflow:>app import --uri https://dataflow.spring.io/rabbitmq-maven-latest

2. Get the PCC connection information

   $ cf service-key cloudcache my-service-key
   Getting key my-service-key for service instance cloudcache as <user>...
   
   {
    "locators": [
     "10.0.16.9[55221]",
     "10.0.16.11[55221]",
     "10.0.16.10[55221]"
    ],
    "urls": {
     "gfsh": "http://...",
     "pulse": "https://.../pulse"
    },
    "users": [
     {
      "password": <password>,
      "username": "cluster_operator"
     },
     {
      "password": <password>,
      "username": "developer"
     }
    ]
   }

3. Using gfsh, connect to the PCC instance as cluster_operator using the service key values and create the Test region.

   gfsh>connect --use-http --url=<gfsh-url> --user=cluster_operator --password=<cluster_operator_password>
   gfsh>create region --name Test --type=REPLICATE

4. Create the stream, connecting to the PCC instance as developer. This example creates an gemfire source to which will publish events on a region

   dataflow>stream create --name events --definition " gemfire --username=developer --password=<developer-password> --connect-type=locator --host-addresses=10.0.16.9:55221 --regionName=Test | log" --deploy

5. Verify the stream is successfully deployed

6. Monitor stdout for the log sink

   cf logs <log-sink-app-name>

7. Using gfsh, create and update some cache entries

   gfsh>connect --use-http --url=<gfsh-url> --user=cluster_operator --password=<cluster_operator_password>
   gfsh>put --region /Test --key 1  --value "value 1"
   gfsh>put --region /Test --key 2  --value "value 2"
   gfsh>put --region /Test --key 3  --value "value 3"
   gfsh>put --region /Test --key 1  --value "new value 1"

8. Observe the log output

   You should see messages like:

   2017-10-28 17:28:52.893  INFO 18986 --- [emfire.events-1] log sink                               : value 1"
   2017-10-28 17:28:52.893  INFO 18986 --- [emfire.events-1] log sink                               : value 2"
   2017-10-28 17:28:52.893  INFO 18986 --- [emfire.events-1] log sink                               : value 3"
   2017-10-28 17:28:52.893  INFO 18986 --- [emfire.events-1] log sink                               : new value 1"

   By default, the message payload contains the updated value. Depending on your application, you may need additional information. The data comes from EntryEvent. You can access any fields using the source’s cache-event-expression property. This takes a SpEL expression bound to the EntryEvent. Try something like --cache-event-expression='{key:'+key+',new_value:'+newValue+'}' (HINT: You will need to destroy the stream and recreate it to add this property, an exercise left to the reader). Now you should see log messages like:

   2017-10-28 17:28:52.893  INFO 18986 --- [emfire.events-1] log-sink                                 : {key:1,new_value:value 1}
   2017-10-28 17:41:24.466  INFO 18986 --- [emfire.events-1] log-sink                                 : {key:2,new_value:value 2}

9. You’re done!