Angular + Spring WebFlux + Spring Data Reactive Cassandra example
Tags: angular, springwebflux, springdata, reactive
Published: 2021-03-16 11:25:10

https://grokonez.com/reactive-programming/angular-spring-webflux-spring-data-reactive-cassandra-example-full-reactive-angular-http-client-spring-boot-restapi-server

In this tutorial, we're gonna build a full Reactive Application in which, Spring WebFlux, Spring Data Reactive Cassandra are used for backend, and Angular, RxJS, EventSource are on client side.

Related Posts:
- <a href="https://grokonez.com/spring-framework/spring-boot/springdata-reactive-cassandra-repositories-springboot">SpringData Reactive Cassandra Repositories | SpringBoot</a>
- <a href="https://grokonez.com/frontend/introduction-to-rxjs-javascript-reactive-stream">Introduction to RxJS – Extensions for JavaScript Reactive Streams</a>
- <a href="https://grokonez.com/spring-framework/spring-boot/angular-4-spring-boot-cassandra-crud-example">Angular 4 + Spring Boot + Cassandra CRUD example</a>

<!--more-->
<h2>I. Technologies</h2>
- Java 1.8
- Maven 3.3.9
- Spring Tool Suite 3.9.0.RELEASE
- Spring Boot 2.0.0.RELEASE
- Angular 5
- RxJS 5.1.0
- Cassandra 3.9.0
<h2>II. Overview</h2>
<h3>1. Full Stack Architecture</h3>
<img src="https://grokonez.com/wp-content/uploads/2018/03/angular-spring-webflux-spring-data-cassandra-reactive-architecture.png" alt="angular-spring-webflux-spring-data-cassandra-reactive-architecture" width="700" height="226" class="alignnone size-full wp-image-11213" />
<h3>2. Reactive Spring Boot Server</h3>
<h5>2.1 Dependency</h5>
<pre><code class="language-html">&lt;dependency&gt;
	&lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
	&lt;artifactId&gt;spring-boot-starter-data-cassandra-reactive&lt;/artifactId&gt;
&lt;/dependency&gt;

&lt;dependency&gt;
	&lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
	&lt;artifactId&gt;spring-boot-starter-webflux&lt;/artifactId&gt;
&lt;/dependency&gt;</code></pre>
<h5>2.2 Reactive Repository</h5>
We just need to create an interface that extends <code>ReactiveCrudRepository</code> to do CRUD operations for a specific type. This repository follows reactive paradigms and uses Project Reactor types (<code>Flux</code>, <code>Mono</code>) which are built on top of Reactive Streams.
<pre><code class="language-java">
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import reactor.core.publisher.Flux;

public interface ReactiveCustomerRepository extends ReactiveCrudRepository<Customer, UUID>{
	Flux<Customer> findByLastname(String lastname);

	@Query("SELECT * FROM customer WHERE firstname = ?0 and lastname  = ?1")
	Mono<Customer> findByFirstnameAndLastname(String firstname, String lastname);

	// for deferred execution
	Flux<Customer> findByLastname(Mono<String> lastname);

	Mono<Customer> findByFirstnameAndLastname(Mono<String> firstname, String lastname);
}</code></pre>
<h5>2.3 Activate reactive Spring Data Cassandra</h5>
Support for reactive Spring Data is activated through an <code>@EnableReactiveCassandraRepositories</code> annotation:
<pre><code class="language-java">
import org.springframework.context.annotation.Configuration;
import org.springframework.data.cassandra.config.AbstractReactiveCassandraConfiguration;
import org.springframework.data.cassandra.config.SchemaAction;
import org.springframework.data.cassandra.repository.config.EnableReactiveCassandraRepositories;

@Configuration
@EnableReactiveCassandraRepositories
public class CassandraReactiveConfig extends AbstractReactiveCassandraConfiguration {

	@Override
	protected String getKeyspaceName() {
		return "javasampleapproach";
	}

	@Override
	public SchemaAction getSchemaAction() {
		return SchemaAction.RECREATE;
	}
}
</code></pre>
<h5>2.4 Call Reactive Repository</h5>
We can forward the reactive parameters provided by Spring Web Reactive, pipe them into the repository, get back a <code>Flux</code>/<code>Mono</code> and then work with result in reactive way.
<pre><code class="language-java">
import java.util.UUID;
import com.datastax.driver.core.utils.UUIDs;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@CrossOrigin(origins = "http://localhost:4200")
@RestController
public class CustomerController {

	@Autowired
	ReactiveCustomerRepository customerRepository;

	@GetMapping("/customers")
	public Flux<Customer> getAllCustomers() {
		return customerRepository.findAll();
	}

	@PostMapping("/customers/create")
	public Mono<Customer> createCustomer(@Valid @RequestBody Customer customer) {
		customer.setId(UUIDs.timeBased());
		customer.setActive(false);
		return customerRepository.save(customer);
	}

	@PutMapping("/customers/{id}")
	public Mono<ResponseEntity<Customer>> updateCustomer(@PathVariable("id") UUID id, @RequestBody Customer customer) {
		return customerRepository.findById(id).flatMap(customerData -> {
			customerData.setName(customer.getName());
			customerData.setAge(customer.getAge());
			customerData.setActive(customer.isActive());
			return customerRepository.save(customerData);
		}).map(updatedcustomer -> new ResponseEntity<>(updatedcustomer, HttpStatus.OK))
				.defaultIfEmpty(new ResponseEntity<>(HttpStatus.NOT_FOUND));
	}

	@DeleteMapping("/customers/{id}")
	public ResponseEntity<String> deleteCustomer(@PathVariable("id") UUID id) {
		try {
			customerRepository.deleteById(id).subscribe();
		} catch (Exception e) {
			return new ResponseEntity<>("Fail to delete!", HttpStatus.EXPECTATION_FAILED);
		}
		return new ResponseEntity<>("Customer has been deleted!", HttpStatus.OK);
	}

	@DeleteMapping("/customers/delete")
	public ResponseEntity<String> deleteAllCustomers() {
		try {
			customerRepository.deleteAll().subscribe();
		} catch (Exception e) {
			return new ResponseEntity<>("Fail to delete!", HttpStatus.EXPECTATION_FAILED);
		}
		return new ResponseEntity<>("All customers have been deleted!", HttpStatus.OK);
	}

	@GetMapping("/customers/findbyage")
	public Flux<Customer> findByAge(@RequestParam int age) {
		return customerRepository.findByAge(age);
	}
}
</code></pre>
In the rest controller methods which are annotated by <code>@RequestMapping</code>, we have used some methods of autowired repository which are implemented interface <code>ReactiveCrudRepository</code>:
<pre><code class="language-java">public interface ReactiveCrudRepository&lt;T, ID&gt; extends Repository&lt;T, ID&gt; {

	&lt;S extends T&gt; Mono&lt;S&gt; save(S entity);
	Mono&lt;T&gt; findById(ID id);
	Flux&lt;T&gt; findAll();
	Mono&lt;Void&gt; deleteById(ID id);
	Mono&lt;Void&gt; deleteAll();
	// ...
}</code></pre>
And <code>findByAge</code> method that we create in our interface ReactiveCustomerRepository:
<pre><code class="language-java">public interface ReactiveCustomerRepository extends ReactiveCrudRepository&lt;Customer, UUID&gt;{
	Flux&lt;Customer&gt; findByAge(int age);
}</code></pre>
Remember that we want to connect to backend from a client application deployed in a different port, so we must enable <em><strong>CORS</strong></em> using <code>@CrossOrigin</code> annotation.
<h3>3. Reactive Angular Client</h3>
<h5>3.1 Reactive Service</h5>
This service interacts with the backend using <em>Server-Sent Events</em>.
<pre><code class="language-javascript">import { Injectable, NgZone } from '@angular/core';
import { HttpClient, HttpRequest } from '@angular/common/http';
import { Observable } from 'rxjs/Observable';
import * as EventSource from 'eventsource';

import { Customer } from './customer';

@Injectable()
export class CustomerService {

  private baseUrl = 'http://localhost:8080/api/customers';
  private customersList: Customer[] = new Array();
  private customersListSearch: Customer[] = new Array();

  constructor(private http: HttpClient, private zone: NgZone) {
  }

  createCustomer(customer: Object): Observable&lt;Object&gt; {
    return this.http.post(`${this.baseUrl}` + `/create`, customer);
  }

  updateCustomer(id: string, value: any): Observable&lt;Object&gt; {
    return this.http.put(`${this.baseUrl}/${id}`, value);
  }

  deleteCustomer(id: string): Observable&lt;any&gt; {
    return this.http.delete(`${this.baseUrl}/${id}`, { responseType: 'text' });
  }

  getCustomersList(): Observable&lt;any&gt; {
    this.customersList = new Array();

    return Observable.create((observer) =&gt; {
      const eventSource = new EventSource(`${this.baseUrl}`);

      eventSource.onmessage = (event) =&gt;
        this.zone.run(() =&gt; {
          console.log('eventSource.onmessage: ', event);
          const json = JSON.parse(event.data);
          this.customersList.push(new Customer(json['id'], json['name'], json['age'], json['active']));
          observer.next(this.customersList);
        });

      eventSource.onerror = (error) =&gt; observer.error('eventSource.onerror: ' + error);

      return () =&gt; eventSource.close();
    });
  }

  deleteAll(): Observable&lt;any&gt; {
    return this.http.delete(`${this.baseUrl}` + `/delete`, { responseType: 'text' });
  }

  findCustomers(age): Observable&lt;any&gt; {
    this.customersListSearch = new Array();

    return Observable.create((observer) =&gt; {
      const eventSource = new EventSource(`${this.baseUrl}` + `/findbyage?age=` + age);

      eventSource.onmessage = (event) =&gt;
        this.zone.run(() =&gt; {
          console.log('eventSource.onmessage: ', event);
          const json = JSON.parse(event.data);
          this.customersListSearch.push(new Customer(json['id'], json['name'], json['age'], json['active']));
          observer.next(this.customersListSearch);
        });

      eventSource.onerror = (error) =&gt; observer.error('eventSource.onerror: ' + error);

      return () =&gt; eventSource.close();
    });
  }
}</code></pre>

Whenever we receive an <code>event</code> through the <code>EventSource</code> object, <code>onmessage()</code> is invoked. That's where we parse data and update item list.

Using RxJS Observable object, any Observer that subscribed to the Observable we created can receive events when the item list gets updated (when calling <code>observer.next(...)</code>).

We have to instantiate a <code>NgZone</code> and call <code>zone.run(callback)</code> to help Angular notify when change happens.
<div class='jsa-quote'>
For more details about RxJS, please visit:
<a href="https://grokonez.com/frontend/introduction-to-rxjs-javascript-reactive-stream">Introduction to RxJS – Extensions for JavaScript Reactive Streams</a>
</div>
<h5>3.2 Reactive Component</h5>
This Component calls Service above and keep result inside an <code>Observable</code> object:
<pre><code class="language-java">
import { Component, OnInit } from '@angular/core';
import { Observable } from 'rxjs/Observable';

import { CustomerService } from '../customer.service';
import { Customer } from '../customer';

@Component({
  selector: 'customers-list',
  templateUrl: './customers-list.component.html',
  styleUrls: ['./customers-list.component.css']
})
export class CustomersListComponent implements OnInit {

  customers: Observable<Customer[]>;

  constructor(private customerService: CustomerService) {  }

  ngOnInit() {
    this.reloadData();
  }

  deleteCustomers() {
    this.customerService.deleteAll()
      .subscribe(
        data => console.log(data),
        error => console.log('ERROR: ' + error)
      );
  }

  reloadData() {
    this.customers = this.customerService.getCustomersList();
  }
}</code></pre>
In HTML template, we add <code>async</code> pipe that subscribes to the Observable and update component whenever a new event comes:
<pre><code class="language-html">&lt;div *ngFor=&quot;let customer of customers | async&quot;&gt;
	&lt;customer-details [customer]='customer'&gt;&lt;/customer-details&gt;
&lt;/div&gt;

&lt;div&gt;
	&lt;button type=&quot;button&quot; class=&quot;button btn-danger&quot; (click)='deleteCustomers()'&gt;Delete All&lt;/button&gt;
&lt;/div&gt;</code></pre>
<h2>III. Practice</h2>
<h3>0. Set up Cassandra</h3>
Open <strong>Cassandra CQL Shell</strong>:

- Create Cassandra <strong>keyspace</strong> with name <strong>javasampleapproach</strong>:
<pre><code class="language-javascript">
create keyspace javasampleapproach with replication={'class':'SimpleStrategy', 'replication_factor':1};
</code></pre>

- Create <strong>customer</strong> table for <strong>javasampleapproach</strong> keyspace:
<pre><code class="language-javascript">
use javasampleapproach;

CREATE TABLE customer(
   id timeuuid PRIMARY KEY,
   name text,
   age int,
   active boolean
);</code></pre>

- Because we use repository finder method on <code>age</code> field, so we need to create an index on <code>age</code> column:
<pre><code class="language-javascript">
CREATE INDEX ON javasampleapproach.customer (age);
</code></pre>
<h3>1. Reactive Spring Boot Server</h3>
<h5>1.1 Project Structure</h5>
<img src="https://grokonez.com/wp-content/uploads/2018/03/angular-spring-webflux-spring-data-cassandra-reactive-server-structure.png" alt="angular-spring-webflux-spring-data-cassandra-reactive-server-structure" width="347" height="381" class="alignnone size-full wp-image-11216" />

- Class <strong>Customer</strong> corresponds to document in <strong>customer</strong> collection.
- <strong>ReactiveCustomerRepository</strong> is an interface extends <strong>ReactiveCrudRepository</strong>, will be autowired in <strong>CustomerController</strong> for implementing repository methods.
- <strong>CustomerController</strong> is a REST Controller which has request mapping methods for RESTful requests such as: <code>getAll</code>, <code>create</code>, <code>update</code>, <code>delete</code> Customers.
- Dependencies for <strong>Spring Boot WebFlux</strong> and <strong>Spring Data Cassandra</strong> in <strong>pom.xml</strong>.
<h5>1.2 Dependency</h5>
<pre><code class="language-html">&lt;dependency&gt;
	&lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
	&lt;artifactId&gt;spring-boot-starter-data-cassandra-reactive&lt;/artifactId&gt;
&lt;/dependency&gt;

&lt;dependency&gt;
	&lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;
	&lt;artifactId&gt;spring-boot-starter-webflux&lt;/artifactId&gt;
&lt;/dependency&gt;</code></pre>
<h5>1.3 Data Model</h5>
<pre><code class="language-java">
package com.javasampleapproach.reactive.cassandra.model;

import java.util.UUID;

import org.springframework.data.cassandra.core.mapping.PrimaryKey;
import org.springframework.data.cassandra.core.mapping.Table;

@Table
public class Customer {
 
	@PrimaryKey
	private UUID id;
 
	private String name;
	private int age;
	private boolean active;
 
	public Customer() {
	}
 
	public Customer(String name, int age) {
		this.name = name;
		this.age = age;
	}
 
	public UUID getId() {
		return id;
	}
 
	public void setId(UUID id) {
		this.id = id;
	}
 
	public void setName(String name) {
		this.name = name;
	}
 
	public String getName() {
		return this.name;
	}
 
	public void setAge(int age) {
		this.age = age;
	}
 
	public int getAge() {
		return this.age;
	}
 
	public boolean isActive() {
		return active;
	}
 
	public void setActive(boolean active) {
		this.active = active;
	}
 
	@Override
	public String toString() {
		return "Customer [id=" + id + ", name=" + name + ", age=" + age + ", active=" + active + "]";
	}
}
</code></pre>
<h5>1.4 Reactive Repository</h5>
<pre><code class="language-java">
package com.javasampleapproach.reactive.cassandra.repo;

import java.util.UUID;

import org.springframework.data.repository.reactive.ReactiveCrudRepository;

import com.javasampleapproach.reactive.cassandra.model.Customer;

import reactor.core.publisher.Flux;

public interface ReactiveCustomerRepository extends ReactiveCrudRepository<Customer, UUID>{
	Flux<Customer> findByAge(int age);
}
</code></pre>
<h5>1.5 Enable reactive Spring Data Cassandra</h5>
<pre><code class="language-java">
package com.javasampleapproach.reactive.cassandra.config;

import org.springframework.context.annotation.Configuration;
import org.springframework.data.cassandra.config.AbstractReactiveCassandraConfiguration;
import org.springframework.data.cassandra.config.SchemaAction;
import org.springframework.data.cassandra.repository.config.EnableReactiveCassandraRepositories;

@Configuration
@EnableReactiveCassandraRepositories
public class CassandraReactiveConfig extends AbstractReactiveCassandraConfiguration {

	@Override
	protected String getKeyspaceName() {
		return "javasampleapproach";
	}

	@Override
	public SchemaAction getSchemaAction() {
		return SchemaAction.RECREATE;
	}

}
</code></pre>
<h5>1.6 REST Controller</h5>
<pre><code class="language-java">
package com.javasampleapproach.reactive.cassandra.controller;

import java.time.Duration;
import java.util.UUID;

import javax.validation.Valid;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import com.javasampleapproach.reactive.cassandra.model.Customer;
import com.javasampleapproach.reactive.cassandra.repo.ReactiveCustomerRepository;

import com.datastax.driver.core.utils.UUIDs;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@CrossOrigin(origins = "http://localhost:4200")
@RestController
@RequestMapping(value = "/api")
public class CustomerController {

	@Autowired
	ReactiveCustomerRepository customerRepository;

	@GetMapping("/customers")
	public Flux<Customer> getAllCustomers() {
		System.out.println("Get all Customers...");

		return customerRepository.findAll().delayElements(Duration.ofMillis(1000));
	}

	@PostMapping("/customers/create")
	public Mono<Customer> createCustomer(@Valid @RequestBody Customer customer) {
		System.out.println("Create Customer: " + customer.getName() + "...");

		customer.setId(UUIDs.timeBased());
		customer.setActive(false);
		return customerRepository.save(customer);
	}

	@PutMapping("/customers/{id}")
	public Mono<ResponseEntity<Customer>> updateCustomer(@PathVariable("id") UUID id, @RequestBody Customer customer) {
		System.out.println("Update Customer with ID = " + id + "...");

		return customerRepository.findById(id).flatMap(customerData -> {
			customerData.setName(customer.getName());
			customerData.setAge(customer.getAge());
			customerData.setActive(customer.isActive());
			return customerRepository.save(customerData);
		}).map(updatedcustomer -> new ResponseEntity<>(updatedcustomer, HttpStatus.OK))
				.defaultIfEmpty(new ResponseEntity<>(HttpStatus.NOT_FOUND));
	}

	@DeleteMapping("/customers/{id}")
	public ResponseEntity<String> deleteCustomer(@PathVariable("id") UUID id) {
		System.out.println("Delete Customer with ID = " + id + "...");

		try {
			customerRepository.deleteById(id).subscribe();
		} catch (Exception e) {
			return new ResponseEntity<>("Fail to delete!", HttpStatus.EXPECTATION_FAILED);
		}

		return new ResponseEntity<>("Customer has been deleted!", HttpStatus.OK);
	}

	@DeleteMapping("/customers/delete")
	public ResponseEntity<String> deleteAllCustomers() {
		System.out.println("Delete All Customers...");

		try {
			customerRepository.deleteAll().subscribe();
		} catch (Exception e) {
			return new ResponseEntity<>("Fail to delete!", HttpStatus.EXPECTATION_FAILED);
		}

		return new ResponseEntity<>("All customers have been deleted!", HttpStatus.OK);
	}

	@GetMapping("/customers/findbyage")
	public Flux<Customer> findByAge(@RequestParam int age) {

		return customerRepository.findByAge(age).delayElements(Duration.ofMillis(1000));
	}
}
</code></pre>
To make the result live, we use <code>delayElements()</code>. It causes a delayed time between 2 events.
<h4>2. Reactive Angular Client</h4>
<h5>2.1 User Interface</h5>
<img src="https://grokonez.com/wp-content/uploads/2018/03/angular-spring-webflux-spring-data-cassandra-reactive-client-ui.png" alt="angular-spring-webflux-spring-data-cassandra-reactive-client-ui" width="700" height="410" class="alignnone size-full wp-image-11222" />

<h5>2.2 Project Structure</h5>
<img src="https://grokonez.com/wp-content/uploads/2018/03/angular-spring-webflux-spring-data-cassandra-reactive-client-structure.png" alt="angular-spring-webflux-spring-data-cassandra-reactive-client-structure" width="213" height="461" class="alignnone size-full wp-image-11217" />

In this example, we have:
– 4 components: <strong><em>customers-list, customer-details, create-customer, search-customers</strong></em>.
– 4 modules: <strong><em>FormsModule, HttpClientModule, AppRoutingModule</strong></em>.
– <strong><em>customer.ts</strong></em>: class Customer (id, name, age, active).
– <strong><em>customer.service.ts</em></strong>: Service for <code>HttpClient</code> methods.
<h5>2.3 AppModule</h5>
<strong>app.module.ts</strong>
<pre><code class="language-java">
import { AppRoutingModule } from './app-routing.module';
import { BrowserModule } from '@angular/platform-browser';
import { NgModule } from '@angular/core';
import { FormsModule } from '@angular/forms';
import { HttpClientModule } from '@angular/common/http';

import { AppComponent } from './app.component';
import { CreateCustomerComponent } from './customers/create-customer/create-customer.component';
import { CustomerDetailsComponent } from './customers/customer-details/customer-details.component';
import { CustomersListComponent } from './customers/customers-list/customers-list.component';
import { SearchCustomersComponent } from './customers/search-customers/search-customers.component';

import { CustomerService } from './customers/customer.service';

@NgModule({
  declarations: [
    AppComponent,
    CreateCustomerComponent,
    CustomerDetailsComponent,
    CustomersListComponent,
    SearchCustomersComponent
  ],
  imports: [
    BrowserModule,
    FormsModule,
    AppRoutingModule,
    HttpClientModule
  ],
  providers: [CustomerService],
  bootstrap: [AppComponent]
})
export class AppModule { }
</code></pre>
<h5>2.4 Model</h5>
<strong>customer.ts</strong>
<pre><code class="language-java">
export class Customer {
  id: string;
  name: string;
  age: number;
  active: boolean;

  constructor(id?: string, name?: string, age?: number, active?: boolean) {
    this.id = id;
    this.name = name;
    this.age = age;
    this.active = active;
  }
}
</code></pre>
<h5>2.5 Service</h5>
<strong>customer.service.ts</strong>
<pre><code class="language-javascript">import { Injectable, NgZone } from '@angular/core';
import { HttpClient, HttpRequest } from '@angular/common/http';
import { Observable } from 'rxjs/Observable';
import * as EventSource from 'eventsource';

import { Customer } from './customer';

@Injectable()
export class CustomerService {

  private baseUrl = 'http://localhost:8080/api/customers';
  private customersList: Customer[] = new Array();
  private customersListSearch: Customer[] = new Array();

  constructor(private http: HttpClient, private zone: NgZone) {
  }

  createCustomer(customer: Object): Observable&lt;Object&gt; {
    return this.http.post(`${this.baseUrl}` + `/create`, customer);
  }

  updateCustomer(id: string, value: any): Observable&lt;Object&gt; {
    return this.http.put(`${this.baseUrl}/${id}`, value);
  }

  deleteCustomer(id: string): Observable&lt;any&gt; {
    return this.http.delete(`${this.baseUrl}/${id}`, { responseType: 'text' });
  }

  getCustomersList(): Observable&lt;any&gt; {
    this.customersList = new Array();

    return Observable.create((observer) =&gt; {
      const eventSource = new EventSource(`${this.baseUrl}`);

      eventSource.onmessage = (event) =&gt;
        this.zone.run(() =&gt; {
          console.log('eventSource.onmessage: ', event);
          const json = JSON.parse(event.data);
          this.customersList.push(new Customer(json['id'], json['name'], json['age'], json['active']));
          observer.next(this.customersList);
        });

      eventSource.onerror = (error) =&gt; observer.error('eventSource.onerror: ' + error);

      return () =&gt; eventSource.close();
    });
  }

  deleteAll(): Observable&lt;any&gt; {
    return this.http.delete(`${this.baseUrl}` + `/delete`, { responseType: 'text' });
  }

  findCustomers(age): Observable&lt;any&gt; {
    this.customersListSearch = new Array();

    return Observable.create((observer) =&gt; {
      const eventSource = new EventSource(`${this.baseUrl}` + `/findbyage?age=` + age);

      eventSource.onmessage = (event) =&gt;
        this.zone.run(() =&gt; {
          console.log('eventSource.onmessage: ', event);
          const json = JSON.parse(event.data);
          this.customersListSearch.push(new Customer(json['id'], json['name'], json['age'], json['active']));
          observer.next(this.customersListSearch);
        });

      eventSource.onerror = (error) =&gt; observer.error('eventSource.onerror: ' + error);

      return () =&gt; eventSource.close();
    });
  }
}
</code></pre>
<h5>2.6 Components</h5>
<h6>2.6.1 CustomerDetailsComponent</h6>
<strong>customer-details.component.ts</strong>
<pre><code class="language-java">
import { Component, OnInit, Input } from '@angular/core';

import { CustomerService } from '../customer.service';
import { Customer } from '../customer';

import { CustomersListComponent } from '../customers-list/customers-list.component';

@Component({
  selector: 'customer-details',
  templateUrl: './customer-details.component.html',
  styleUrls: ['./customer-details.component.css']
})
export class CustomerDetailsComponent implements OnInit {

  @Input() customer: Customer;

  constructor(private customerService: CustomerService, private listComponent: CustomersListComponent) { }

  ngOnInit() {
  }

  updateActive(isActive: boolean) {
    this.customerService.updateCustomer(this.customer.id,
      { name: this.customer.name, age: this.customer.age, active: isActive })
      .subscribe(
        data => {
          console.log(data);
          this.customer = data as Customer;
        },
        error => console.log(error)
      );
  }

  deleteCustomer() {
    this.customerService.deleteCustomer(this.customer.id)
      .subscribe(
        data => {
          console.log(data);
          this.listComponent.reloadData();
        },
        error => console.log(error)
      );
  }

}
</code></pre>
<strong>customer-details.component.html</strong>
<pre><code class="language-html">&lt;div *ngIf=&quot;customer&quot;&gt;
	&lt;div&gt;
		&lt;label&gt;Name: &lt;/label&gt;  { { customer.name}}
	&lt;/div&gt;
	&lt;div&gt;
		&lt;label&gt;Age: &lt;/label&gt;  { { customer.age}}
	&lt;/div&gt;
	&lt;div&gt;
		&lt;label&gt;Active: &lt;/label&gt;  { { customer.active}}
	&lt;/div&gt;

	&lt;span class=&quot;button is-small btn-primary&quot; *ngIf='customer.active' (click)='updateActive(false)'&gt;Inactive&lt;/span&gt;

	&lt;span class=&quot;button is-small btn-primary&quot; *ngIf='!customer.active' (click)='updateActive(true)'&gt;Active&lt;/span&gt;

	&lt;span class=&quot;button is-small btn-danger&quot; (click)='deleteCustomer()'&gt;Delete&lt;/span&gt;

	&lt;hr/&gt;
&lt;/div&gt;</code></pre>
<h6>2.6.2 CustomersListComponent</h6>
<strong>customers-list.component.ts</strong>
<pre><code class="language-java">
import { Component, OnInit } from '@angular/core';
import { Router } from '@angular/router';
import { Observable } from 'rxjs/Observable';

import { CustomerService } from '../customer.service';
import { Customer } from '../customer';

@Component({
  selector: 'customers-list',
  templateUrl: './customers-list.component.html',
  styleUrls: ['./customers-list.component.css']
})
export class CustomersListComponent implements OnInit {

  customers: Observable<Customer[]>;

  constructor(private customerService: CustomerService, private router: Router) { }

  ngOnInit() {
    this.reloadData();
  }

  deleteCustomers() {
    this.customerService.deleteAll()
      .subscribe(
        data => {
          console.log(data);
          this.navigateToAdd();
        },
        error => console.log('ERROR: ' + error)
      );
  }

  reloadData() {
    this.customers = this.customerService.getCustomersList();
  }

  navigateToAdd() {
    this.router.navigate(['add']);
  }
}</code></pre>
<strong>customers-list.component.html</strong>
<pre><code class="language-html">&lt;br/&gt;
&lt;div *ngFor=&quot;let customer of customers | async&quot; style=&quot;width: 300px;&quot;&gt;
	&lt;customer-details [customer]='customer'&gt;&lt;/customer-details&gt;
&lt;/div&gt;

&lt;div&gt;
	&lt;button type=&quot;button&quot; class=&quot;button btn-danger&quot; (click)='deleteCustomers()'&gt;Delete All&lt;/button&gt;
&lt;/div&gt;</code></pre>
<h6>2.6.3 CreateCustomerComponent</h6>
<strong>create-customer.component.ts</strong>
<pre><code class="language-javascript">
import { Component, OnInit } from '@angular/core';
import { FormsModule } from '@angular/forms';

import { Customer } from '../customer';
import { CustomerService } from '../customer.service';

@Component({
  selector: 'create-customer',
  templateUrl: './create-customer.component.html',
  styleUrls: ['./create-customer.component.css']
})
export class CreateCustomerComponent implements OnInit {

  customer: Customer = new Customer();
  submitted = false;

  constructor(private customerService: CustomerService) { }

  ngOnInit() {
  }

  newCustomer(): void {
    this.submitted = false;
    this.customer = new Customer();
  }

  save() {
    this.customerService.createCustomer(this.customer)
      .subscribe(data => console.log(data), error => console.log(error));
    this.customer = new Customer();
  }

  onSubmit() {
    this.submitted = true;
    this.save();
  }

}
</code></pre>
<strong>create-customer.component.html</strong>
<pre><code class="language-html">&lt;h3&gt;Create Customer&lt;/h3&gt;
&lt;div [hidden]=&quot;submitted&quot; style=&quot;width: 300px;&quot;&gt;
	&lt;form (ngSubmit)=&quot;onSubmit()&quot;&gt;
		&lt;div class=&quot;form-group&quot;&gt;
			&lt;label for=&quot;name&quot;&gt;Name&lt;/label&gt; &lt;input type=&quot;text&quot;
				class=&quot;form-control&quot; id=&quot;name&quot; required [(ngModel)]=&quot;customer.name&quot;
				name=&quot;name&quot;&gt;
		&lt;/div&gt;

		&lt;div class=&quot;form-group&quot;&gt;
			&lt;label for=&quot;age&quot;&gt;Age&lt;/label&gt; &lt;input type=&quot;text&quot;
				class=&quot;form-control&quot; id=&quot;age&quot; required [(ngModel)]=&quot;customer.age&quot;
				name=&quot;age&quot;&gt;
		&lt;/div&gt;

		&lt;button type=&quot;submit&quot; class=&quot;btn btn-success&quot;&gt;Submit&lt;/button&gt;
	&lt;/form&gt;
&lt;/div&gt;

&lt;div [hidden]=&quot;!submitted&quot;&gt;
	&lt;h4&gt;You submitted successfully!&lt;/h4&gt;
	&lt;button class=&quot;btn btn-success&quot; (click)=&quot;newCustomer()&quot;&gt;Add&lt;/button&gt;
&lt;/div&gt;</code></pre>
<h6>2.6.4 SearchCustomersComponent </h6>
<strong>search-customers.component.ts</strong>
<pre><code class="language-javascript">
import { Component, OnInit } from '@angular/core';
import { Observable } from 'rxjs/Observable';

import { CustomerService } from '../customer.service';
import { Customer } from '../customer';

@Component({
  selector: 'search-customers',
  templateUrl: './search-customers.component.html',
  styleUrls: ['./search-customers.component.css']
})
export class SearchCustomersComponent implements OnInit {

  customers: Observable<Customer[]>;
  age: number;

  constructor(private customerService: CustomerService) { }

  ngOnInit() {
    this.age = 0;
  }

  search() {
    this.customers = this.customerService.findCustomers(this.age);
  }

}</code></pre>
<strong>search-customers.component.html</strong>
<pre><code class="language-html">&lt;h3&gt;Find Customers By Age&lt;/h3&gt;
&lt;input type=&quot;text&quot; [(ngModel)]=&quot;age&quot; placeholder=&quot;enter age&quot; class=&quot;input&quot;&gt;
 
&lt;button class=&quot;btn btn-success&quot; (click)=&quot;search()&quot;&gt;Search&lt;/button&gt;
&lt;hr /&gt;
&lt;ul&gt;
	&lt;li *ngFor=&quot;let customer of customers | async&quot;&gt;
		&lt;h5&gt; { { customer.name}} - Age:  { { customer.age}} - Active:	 { { customer.active}}&lt;/h5&gt;
	&lt;/li&gt;
&lt;/ul&gt;</code></pre>
<h5>2.7 AppRoutingModule</h5>
<strong>app-routing.module.ts</strong>
<pre><code class="language-javascript">
import { CreateCustomerComponent } from './customers/create-customer/create-customer.component';
import { CustomersListComponent } from './customers/customers-list/customers-list.component';
import { SearchCustomersComponent } from './customers/search-customers/search-customers.component';

import { NgModule } from '@angular/core';
import { RouterModule, Routes } from '@angular/router';

const routes: Routes = [
  { path: '', redirectTo: 'customers', pathMatch: 'full' },
  { path: 'customers', component: CustomersListComponent },
  { path: 'add', component: CreateCustomerComponent },
  { path: 'search', component: SearchCustomersComponent },
];

@NgModule({
  imports: [RouterModule.forRoot(routes)],
  exports: [RouterModule]
})

export class AppRoutingModule { }
</code></pre>
<h5>2.8 App Component</h5>
<strong>app.component.ts</strong>
<pre><code class="language-javascript">
import { Component } from '@angular/core';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent {
  title = 'JavaSampleApproach';
  description = 'Reactive-Angular-Cassandra';

  constructor() { }
}</code></pre>
<strong>app.component.html</strong>
<pre><code class="language-html">&lt;div class=&quot;container-fluid&quot;&gt;
  &lt;div style=&quot;color: blue;&quot;&gt;
    &lt;h1&gt; { { title}}&lt;/h1&gt;
    &lt;h3&gt; { { description}}&lt;/h3&gt;
  &lt;/div&gt;

  &lt;nav&gt;
    &lt;a routerLink=&quot;customers&quot; class=&quot;btn btn-primary active&quot; role=&quot;button&quot; routerLinkActive=&quot;active&quot;&gt;Customers&lt;/a&gt;
    &lt;a routerLink=&quot;add&quot; class=&quot;btn btn-primary active&quot; role=&quot;button&quot; routerLinkActive=&quot;active&quot;&gt;Add&lt;/a&gt;
    &lt;a routerLink=&quot;search&quot; class=&quot;btn btn-primary active&quot; role=&quot;button&quot; routerLinkActive=&quot;active&quot;&gt;Search&lt;/a&gt;
  &lt;/nav&gt;
  &lt;router-outlet&gt;&lt;/router-outlet&gt;
&lt;/div&gt;</code></pre>
<h3>3. Run & Check Result</h3>
- <strong>Build</strong> and <strong>Run</strong> <strong>Spring Boot</strong> project with commandlines: <code>mvn clean install</code> and <code>mvn spring-boot:run</code>.
- Run the <strong>Angular App</strong> with command: <code>npm start</code>.

- Open browser with url <code>http://localhost:4200/</code>, add some Customers.
- Click on <strong>Customers</strong> tab, each Customer displays one after another with 1s delayed time.

<img src="https://grokonez.com/wp-content/uploads/2018/03/angular-spring-webflux-spring-data-cassandra-reactive-client-result.png" alt="angular-spring-webflux-spring-data-cassandra-reactive-client-result" width="524" height="472" class="alignnone size-full wp-image-11221" />

- Click on <strong>Search</strong> tab, search '27', the result shows each Customer one after another with 1s delayed time.

<img src="https://grokonez.com/wp-content/uploads/2018/03/angular-spring-webflux-spring-data-cassandra-reactive-client-result-search.png" alt="angular-spring-webflux-spring-data-cassandra-reactive-client-result-search" width="575" height="190" class="alignnone size-full wp-image-11223" />
<h2>IV. Source Code</h2>

- <a href="https://grokonez.com/wp-content/uploads/2018/03/SpringDataReactiveCassandra.zip">SpringDataReactiveCassandra</a>
- <a href="https://grokonez.com/wp-content/uploads/2018/03/ReactiveAngularCassandra.zip">ReactiveAngularCassandra</a>

https://grokonez.com/reactive-programming/angular-spring-webflux-spring-data-reactive-cassandra-example-full-reactive-angular-http-client-spring-boot-restapi-server