Building reactive real time aircrafts tracker using Reactor!
The Reactive Streams specification is an industry-driven effort to standardize Reactive Programming libraries on the JVM, and more importantly specify how they must behave so that they are interoperable. Implementors include Reactor 3, RxJava 2, Akka Streams, Vert.x and Ratpack.
The reactive streams standard provides the minimal API to support this style of architecture which has being introduced in Java 9.
In this article, we're going to build a real time like aircrafts tracker using Spring boot 2. The complete project code is available on github.
Reactor
Reactor is a Java framework from the Spring Team. It builds directly on Reactive Streams, so there is no need for a bridge. The Reactor IO project provides wrappers around low-level network runtimes like Netty and Aeron. Reactor is a "4th Generation" library according to David Karnok’s Generations of Reactive classification.
Reactor provides two main types of publishers: Flux and Mono. A Flux is a general purpose publisher that can contain an unbounded number of events. A Mono is a specialized publisher that can contain only zero or one events.
Building Aircrafts tracker
To better understand the benefits of the reactor-based approach, let’s look at a practical example.
We’re going to build a simple aircraft tracker app, which would track many aircrafts based on data collected from opensky-network.org API, and show aircraft details from the database. A typical synchronous implementation would naturally be bound by the throughput of the API, with many periodic calls to get data. With a reactive approach, the system can be more flexible and adapt better to failures or timeouts.
Let’s have a look at the application, starting with the more traditional aspects and moving on to the more reactive constructs.
POJOs
For this example, we've 3 Plain old java object classes: 2 of them are related to opensky-network.org rest API, which are Flight and StateVector. The third one represents Aircraft info details.
@Data
public class Flight {
private int time;
private Collection<StateVector> states;
}
@Data
@JsonFormat(shape=JsonFormat.Shape.ARRAY)
public class StateVector {
private String icao24;
private String callsign;
private String originCountry;
private Double lastPositionUpdate;
private Double lastContact;
private Double longitude;
private Double latitude;
private Double geoAltitude;
private boolean onGround;
private Double velocity;
private Double heading;
private Double verticalRate;
private Set<Integer> serials;
private Double baroAltitude;
private String squawk;
private boolean spi;
private PositionSource positionSource;
}
@Data
@AllArgsConstructor
@Document(collection="aircraft")
public class Aircraft {
@Id
private String icao;
private String registration, manufacturericao, manufacturername, model, owner, operator, reguntil, engines, built;
}
Repository
I'm using mongodb to store Aircraft information, which means I'm using ReactiveCrudRepository. As you can see below, there’s not too much difference to what we used to. However, in contrast to the traditional repository interfaces, a reactive repository uses reactive types as return types and can do so for parameter types too. The CRUD methods in the ReactiveCrudRepository, of course, make use of these types, too:
public interface AircraftRepository extends ReactiveCrudRepository<Aircraft, Long> {
Mono<Aircraft> findByIcao(String icao);
}
opensky-network.org published a publicly registred aircrafts database in csv format that we need to insert in our mongo database. For that I'm using apache commons csv to parse csv file, filter empty lines and insert registered aircrafts list on my database:
repository.saveAll(aircrafts).subscribe(
v -> LOGGER.info("saving {}", v),
e -> LOGGER.error("Saving Failed", e),
() -> LOGGER.info("Loading Data Complete!")
);
Service Layer
Our FlightService contains 2 main methods. the first one, getAllFlights() make use of the new Spring WebFlux. It provides support for both creating reactive, server-based web applications and also has client libraries to make remote REST calls.
the getFlightDetail(String icao24) simply call the repository and return Mono<Aircraft> details from database.
@Service
public class FlightService {
@Value("${opensky.base_url}")
private String baseURL;
@Value("${opensky.all_states}")
private String allStates;
@Autowired
AircraftRepository repository;
@Bean
WebClient client(){
return WebClient.create(baseURL);
}
public Mono<Flight> getAllFlights(){
return client().get().uri(allStates).accept(MediaType.APPLICATION_JSON)
.exchange()
.flatMap(cr -> cr.bodyToMono(Flight.class));
}
public Mono<Aircraft> getFlightDetail(String icao24){
return repository.findByIcao(icao24);
}
}
Controller
We difine 3 endpoints in our controller:
/: return the home page/flights: Rest endpoint, which returns the list of all live tracked flights./aircraft/{icao}return the aircraft details based on the providedicao
Note that, using annotation to build Reactive web applications is very similar with the Spring MVC, except the input or return type is difference, we will use Reactor objects such as Mono or Flux.
Another thing to keep in mind, we are using produces with the value text/event-stream. This is an attribute, based on the concept of Server-Send Event, help us send updating content from the server to client. The new list of flights is pushed down to the user whenever it is added.
@Controller
public class FlightController {
@Autowired
FlightService service;
@GetMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE, value = "/flights")
@ResponseBody
Flux<Flight> flights(){
return Flux.from(service.getAllFlights());
}
@GetMapping(produces = MediaType.APPLICATION_JSON_VALUE, value = "/aircraft/{icao}")
@ResponseBody
Mono<Aircraft> aircraft(@PathVariable String icao){
return service.getFlightDetail(icao);
}
@GetMapping("/")
Mono<String> home() {
return Mono.just("flights");
}
}
Some Javascript
The very last part is the javascript code, used to read and handle stream from the backend, update markers and information of aircrafts.
PS: This of course can be more optimised ;)
var evtSource = new EventSource("http://localhost:8080/flights");
var map = new google.maps.Map(document.getElementById('map_canvas'), {
zoom: 8,
center: new google.maps.LatLng(37.77, -122.43),
mapTypeId: google.maps.MapTypeId.ROADMAP
});
var infowindow = new google.maps.InfoWindow();
var marker;
//Used to remember markers
var markerStore = {};
// Calling main function
fetchdata();
function convertTimestamp(timestamp) {
var d = new Date(timestamp * 1000), // Convert the passed timestamp to milliseconds
yyyy = d.getFullYear(),
mm = ('0' + (d.getMonth() + 1)).slice(-2), // Months are zero based. Add leading 0.
dd = ('0' + d.getDate()).slice(-2), // Add leading 0.
hh = d.getHours(),
h = hh,
min = ('0' + d.getMinutes()).slice(-2), // Add leading 0.
ampm = 'AM',
s= d.getSeconds(),
time;
if (hh > 12) {
h = hh - 12;
ampm = 'PM';
} else if (hh === 12) {
h = 12;
ampm = 'PM';
} else if (hh == 0) {
h = 12;
}
// ie: 2013-02-18, 8:35:45 AM
time = yyyy + '-' + mm + '-' + dd + ', ' + h + ':' + min + ':'+ s + ' ' + ampm;
return time;
}
// main function, it gets data rom /flights endpoint, and update markers and info based on it,
function fetchdata(){
evtSource.onmessage = function (event) {
var state = JSON.parse(event.data);
var data = state.states;
data.filter(function(state){
return state[6] && state[5];
}).forEach(function(state){
if (markerStore.hasOwnProperty(state[0])) {
var m = markerStore[state[0]];
var icon = m.getIcon();
icon.rotation = state[10] - 90;
m.setPosition(new google.maps.LatLng(state[6] , state[5]));
m.setIcon(icon);
} else {
marker = new google.maps.Marker({
map: map,
position: new google.maps.LatLng(state[6], state[5]),
icon: {
path: "m2,106h28l24,30h72l-44,-133h35l80,132h98c21,0 21,34 0,34l-98,0 -80,134h-35l43,-133h-71l-24,30h-28l15,-47",
rotation: state[10]- 90,
scale: 0.05,
fillOpacity: 0.8,
strokeColor: 'gold',
fillColor: 'blue'
}
});
google.maps.event.addListener(marker, 'click', (function (marker) {
return function () {
infowindow.setContent('ICAO 24: '+ state[0]);
infowindow.open(map, marker);
plane_info(state);
aircraft(state[0]);
}
})(marker));
markerStore[state[0]] = marker;
}
});
tracking_count(Object.keys(markerStore).length, convertTimestamp(state.time));
}
}
// currently tracked aircrafts
function tracking_count(track, time) {
document.getElementById('cur-tracking').innerHTML = track;
document.getElementById('last-update').innerHTML = time;
}
// Update Linfe info card
function plane_info(state) {
var content = '<div class="card-content white-text">' +
'<span class="card-title">Live Info: ICAO24 '+state[0]+'</span> ' +
'<p>callsign: ' + handleempty(state[1]) + '</p>' +
'<p>Country: ' + state[2] + '</p>' +
'<p>Velocity: ' + state[9] + ' m/s</p>' +
'<p>Verticale Rate: ' + state[11] + ' m/s</p>' +
'<p>Squawk: ' + handleempty(state[14]) + '</p>' +
'<p>Landed: ' + state[8] + '</p>' +
'<p>Barom Altitude: ' + handleempty(state[13]) + 'm</p>' +
'<p>heading: ' + state[10] + '°</p>' +
'<div class="card-action"> ' +
'<a href="https://opensky-network.org/network/explorer?icao24='+state[0]+'">More info about this flight</a> ' +
'</div>';
document.getElementById('plane-live').innerHTML = content;
}
// update Aircraft info card
function aircraft(icao) {
$.get(
"http://localhost:8080/aircraft/"+ icao ,
function(data) {
content = '<div class="card-content white-text">' +
'<span class="card-title">Aircraft Info: ICAO 24 ' + data.icao + '</span> ' +
'<p>Registration: ' + handleempty(data.registration) + '</p>' +
'<p>manufacturer icao: ' + handleempty(data.manufacturericao) + '</p>' +
'<p>manufacturer name: ' + handleempty(data.manufacturername) + '</p>' +
'<p>model: ' + handleempty(data.model) + '</p>' +
'<p>Owner: ' + handleempty(data.owner) + '</p>' +
'<p>registered until: ' + handleempty(data.reguntil) + '</p>' +
'<p>Built: ' + handleempty(data.built) + '</p>' +
'<p>operator: ' + handleempty(data.operator) + '</p>'+
'<div class="card-action"> ' +
'<a href="https://opensky-network.org/aircraft-profile?icao24='+data.icao+'">More info about this plane</a> ' +
'</div>';
document.getElementById('plane-info').innerHTML = content;
}, "json"
).fail(document.getElementById('plane-info').innerHTML = ' <div class="card-content white-text"><span class="card-title">No record found for ICAO 24 ' + icao + '</span></div> ' );
}
// in case no value, return N/A!
function handleempty(s){
if(!s || s=="") return 'N/A';
return s;
}
The below gif shows this project in action. If you've any suggestion, remark or question, please do not hesitate to comment or PR on github.