Introduction
El Couchbase Móvil Sync Gateway changes feed provides a way to monitor events in a mobile deployment. The feed makes it feasible to write sophisticated business logic. I wrote a tool to help examine and understand the feed. You can read an introduction and description in part one of this two part series. The code also serves as an example of listening to the feed.
The code
I’ve included the major classes from the app code here. This is the first version, so it can use plenty of enhancements. The parameters are all hard-wired. Check the project here on GitHub for updates. You can also find instructions for building, running, and packaging the app there.
JavaFX: The Controller class
JavaFX breaks simple apps into a controller class and a declarative UI. Let’s walk through the controller.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 |
paquete com.Couchbase.mobile; importar com.Couchbase.lite.*; importar com.fasterxml.jackson.núcleo.JsonProcessingException; importar javafx.aplicación.Plataforma; importar javafx.beans.valor.ChangeListener; importar javafx.beans.valor.ObservableValue; importar javafx.colecciones.FXCollections; importar javafx.colecciones.ObservableList; importar javafx.event.ActionEvent; importar javafx.fxml.FXML; importar javafx.scene.control.*; importar javafx.scene.control.TextField; importar Java.yo.IOException; importar Java.net.MalformedURLException; importar Java.net.URL; importar Java.útil.Map; importar estático com.Couchbase.mobile.Runtime.mapper; público clase Controller implementa LiveQuery.ChangeListener, ChangeListener, SGMonitor.ChangesFeedListener, DBService.ReplicationStateListener { privado estático final Cadena SYNC_GATEWAY_HOST = “https://localhost”; privado estático final Cadena SG_PUBLIC_URL = SYNC_GATEWAY_HOST + “:4984/” + DBService.DATABASE; privado estático final Cadena SG_ADMIN_URL = SYNC_GATEWAY_HOST + “:4985/” + DBService.DATABASE; privado estático final Cadena TOGGLE_INACTIVE = “-fx-background-color: #e6555d;”; privado estático final Cadena TOGGLE_ACTIVE = “-fx-background-color: #ade6a6;”; privado estático final Cadena TOGGLE_DISABLED = “-fx-background-color: #555555;”; @FXML privado ListView documentList; privado ObservableList documents = FXCollections.observableArrayList(); @FXML privado TextArea contentsText; @FXML privado TextArea changesFeed; @FXML privado TextField usernameText; @FXML privado TextField passwordText; @FXML privado ToggleButton applyCredentialsBtn; @FXML privado ToggleButton syncBtn; privado DBService service = DBService.getInstance(); privado Database bd = service.getDatabase(); privado SGMonitor changesMonitor; privado LiveQuery liveQuery; |
This first listing shows a bunch of boiler plate code. I implement several listeners for the UI within the class itself to cut down of files. This is for illustration purposes.
El @FXML annotations mark all the fields that the framework will automatically bind to portions of the UI.
Next comes initialization. JavaFX calls this method as part of its standard lifecycle.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
@FXML privado vacío initialize() { documentListInitialize(); documentList.setItems(documents); setState(applyCredentialsBtn, false); setState(syncBtn, false); service.addReplicationStateListener(este); changesMonitor = nuevo SGMonitor(SG_ADMIN_URL, “false”, “true”, “0”, “all_docs”, este); changesMonitor.start(); } privado vacío documentListInitialize() { Consulta consulta = bd.createAllDocumentsQuery(); consulta.setAllDocsMode(Consulta.AllDocsMode.INCLUDE_DELETED); liveQuery = consulta.toLiveQuery(); liveQuery.addChangeListener(este); liveQuery.start(); documentList.getSelectionModel().selectedItemProperty().addListener(este); } |
I’ve broken out the document list initialization into its own routine. The document list gets bound to the documentList variable. In turn documentList will update the UI whenever the item list we pass in changes.
I set up a live query to monitor the client database for any changes. This happens through an “all docs” query. An all docs query doesn’t require an associated view. I set INCLUDE_DELETED so the tool can show what a deleted document looks like in the database.
With the other bindings in place, we just have to update the documents list. We’ll see the live query listener that does that further along.
The next few lines set the initial state of a couple of toggle buttons. I need an extra listener to keep the Sync button consistent with the actual state of the replications. More on this further along in the article.
I wrote a separate class to monitor Sync Gateway. The initialization code finished by creating a new monitor instance and kicking it off.
The next section contains several listeners.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
// LiveQuery.ChangeListener @Override público vacío changed(LiveQuery.ChangeEvent event) { si (event.getSource().es igual a(liveQuery)) { Plataforma.runLater(() -> { QueryEnumerator filas = event.getRows(); documents.clear(); filas.paraCada(queryRow -> documents.añadir(queryRow.getDocumentId())); }); } } |
Here’s the live query listener that gets called whenever the local database changes. I didn’t design the tool for working with massive databases. So, whenever the data changes, I just took the brute force approach of rereading every document. The getRows method returns an enumerator that will index doing just that. JavaFX takes care of updating the UI when documents changes.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
// ListView ChangeListener @Override público vacío changed(ObservableValue observable, Cadena oldId, Cadena newId) { si (nulo == newId) regresar; Map properties = bd.getDocument(newId).getProperties(); Intentar { Cadena json = mapper.writeValueAsString(properties); contentsText.setText(prettyText(json)); } atrapar (JsonProcessingException ex) { ex.printStackTrace(); Dialog.display(ex); } } |
This listener takes care of tracking when a user clicks on an entry in the document list. The entries are the document IDs, so we can use a selection to pull the document directly from the database.
|
1 2 3 4 5 |
// SGMonitor.ChangesFeedListener @Override público vacío onResponse(Cadena body) { changesFeed.appendText(prettyText((Cadena) body)); } |
I used a callback approach to get the results of the changes feed. The interface is defined in the SGMonitor class. It has just the one method. In this implementation I simply take the body of the feed response and tack it on to the existing text in the changes feed text pane. There’s a little formatting done to make it easier to read, too.
|
1 2 3 4 5 |
// DBService.ReplicationStateListener @Override público vacío onChange(boolean isActive) { setState(syncBtn, isActive); } |
Finally I added a listener for replication activity. The interface comes from the DBService helper class. I wrote a bit about detecting the state of a replication Aquí. For this app I just need to know whether a replication is running or not to keep the Sync button state consistent. This handles cases where a user tries to start a sync but it fails. This can happen if they need to provide authentication credentials but haven’t, for example.
Next we have several methods bound to UI elements. JavaFX handles much of the wiring.
|
1 2 3 4 5 6 7 8 9 10 11 12 |
@FXML privado vacío applyCredentialsToggled(ActionEvent event) { Cadena nombre de usuario = nulo; Cadena contraseña = nulo; si (applyCredentialsBtn.isSelected()) { nombre de usuario = usernameText.getText(); contraseña = passwordText.getText(); } DBService.getInstance().setCredentials(nombre de usuario, contraseña); applyCredentialsBtn.setStyle(applyCredentialsBtn.isSelected() ? TOGGLE_ACTIVE : TOGGLE_INACTIVE); } |
Here I set the use of authentication credentials whenever the corresponding button gets toggled.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
@FXML privado vacío saveContentsClicked(ActionEvent event) { Map properties = nulo; Documento documento; Intentar { properties = mapper.readValue(contentsText.getText(), Map.clase); } atrapar (IOException ex) { ex.printStackTrace(); Dialog.display(ex); } si (properties.containsKey(“_id”)) { documento = bd.getDocument((Cadena) properties.conseguir(“_id”)); } otro { documento = bd.createDocument(); } Intentar { documento.putProperties(properties); } atrapar (CouchbaseLiteException ex) { ex.printStackTrace(); Dialog.display(ex); } } |
This code shows a couple of interesting items. I use a Jackson ObjectMapper instance to convert the text in the content pane to a property map.
Next I check for an entry _id. Couchbase Mobile reserves most properties starting with an “_” for system use (with special exceptions). If the text we’re trying to convert contains _id, I assume this is an edit to an existing document. Otherwise I create a new document.
So, in a nutshell, we have an example of both creating and updating documents. This isn’t the preferred way to update, although it suffices in many cases. You can read more about updates Aquí.
|
1 2 3 4 5 6 7 8 9 10 11 |
@FXML privado vacío syncToggled(ActionEvent event) { Intentar { syncBtn.setDisable(true); syncBtn.setStyle(TOGGLE_DISABLED); service.toggleReplication(nuevo URL(SG_PUBLIC_URL), true); } atrapar (Excepción ex) { ex.printStackTrace(); Dialog.display(ex); syncBtn.setDisable(false); } } |
This reacts to toggling the Sync button. Recall though that we use a listener to verify the state elsewhere.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 |
@FXML privado vacío exitClicked(ActionEvent event) { // Try to shut everything down gracefully changesMonitor.stop(); liveQuery.stop(); service.stopReplication(); bd.close(); bd.getManager().close(); Plataforma.exit(); } privado vacío setState(ToggleButton btn, boolean active) { btn.setSelected(active); btn.setStyle(active ? TOGGLE_ACTIVE : TOGGLE_INACTIVE); btn.setDisable(false); } privado Cadena prettyText(Cadena json) { Cadena out = nulo; Intentar { Object objeto = mapper.readValue(json, Object.clase); out = mapper.writerWithDefaultPrettyPrinter().writeValueAsString(objeto); } atrapar (Excepción ex) { ex.printStackTrace(); } regresar out; } } |
The rest of the code here are just helper bits and a piece to shutdown everything before exiting.
The Database Helper class
This shows the code for a straight-forward database helper class. For the most part I just find this class a nice packaging of the typical operations needed for managing a database and starting a standard bidirectional set of replications. I’m including it here because I find it useful and for clarity.
I do implement the Replication.ChangeListener interface. That’s maybe a little unusual. I mentioned the reason earlier on. This link takes you to the entrada de blog about it.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 |
paquete com.Couchbase.mobile; importar com.Couchbase.lite.Database; importar com.Couchbase.lite.JavaContext; importar com.Couchbase.lite.Manager; importar com.Couchbase.lite.auth.Authenticator; importar com.Couchbase.lite.auth.AuthenticatorFactory; importar com.Couchbase.lite.replicator.Replication; importar com.Couchbase.lite.replicator.ReplicationState; importar Java.net.URL; importar Java.útil.ArrayList; importar Java.útil.Lista; público clase DBService implementa Replication.ChangeListener { público estático final Cadena DATABASE = “db”; privado estático final Cadena DB_DIRECTORY = “data”; privado Manager manager; privado Database base de datos; privado Replication pushReplication = nulo; privado Replication pullReplication = nulo; privado boolean replicationActive = false; privado Lista stateListeners = nuevo ArrayList(); privado Cadena nombre de usuario = nulo; privado Cadena contraseña = nulo; privado DBService() { Intentar { manager = nuevo Manager(nuevo JavaContext(DB_DIRECTORY), Manager.DEFAULT_OPTIONS); base de datos = manager.getDatabase(DATABASE); } atrapar (Excepción ex) { ex.printStackTrace(); } } privado estático clase Holder { privado estático DBService INSTANCE = nuevo DBService(); } público interfaz ReplicationStateListener { vacío onChange(boolean isActive); } público estático DBService getInstance() { regresar Holder.INSTANCE; } público Database getDatabase() { regresar base de datos; } público vacío setCredentials(Cadena nombre de usuario, Cadena contraseña) { este.nombre de usuario = nombre de usuario; este.contraseña = contraseña; } público vacío toggleReplication(URL gateway, boolean continuous) { si (replicationActive) { stopReplication(); } otro { startReplication(gateway, continuous); } } público vacío startReplication(URL gateway, boolean continuous) { si (replicationActive) { stopReplication(); } pushReplication = base de datos.createPushReplication(gateway); pullReplication = base de datos.createPullReplication(gateway); pushReplication.setContinuous(continuous); pullReplication.setContinuous(continuous); si (nombre de usuario != nulo) { Authenticator auth = AuthenticatorFactory.createBasicAuthenticator(nombre de usuario, contraseña); pushReplication.setAuthenticator(auth); pullReplication.setAuthenticator(auth); } pushReplication.addChangeListener(este); pullReplication.addChangeListener(este); pushReplication.start(); pullReplication.start(); } público vacío stopReplication() { si (!replicationActive) regresar; pushReplication.stop(); pullReplication.stop(); pushReplication = nulo; pullReplication = nulo; } público vacío addReplicationStateListener(ReplicationStateListener listener) { stateListeners.añadir(listener); } público vacío removeReplicationStateListener(ReplicationStateListener listener) { stateListeners.remove(listener); } // Replication.ChangeListener @Override público vacío changed(Replication.ChangeEvent changeEvent) { si (changeEvent.getError() != nulo) { Throwable lastError = changeEvent.getError(); Dialog.display(lastError.getMessage()); regresar; } si (changeEvent.getTransition() == nulo) regresar; ReplicationState dest = changeEvent.getTransition().getDestination(); replicationActive = ((dest == ReplicationState.STOPPING || dest == ReplicationState.STOPPED) ? false : true); stateListeners.paraCada(listener -> listener.onChange(replicationActive)); } } |
The Sync Gateway Monitor class
Finally, let’s take a look at the helper class for monitoring Sync Gateway. I’ll walk through this in pieces, too.
0
I use the OkHttp library from Square. Currently Couchbase Lite uses this library too, internally. OkHttp uses a builder pattern. I prepare a builder instance I’ll use through the rest of the code in the class constructor. You can read about the meaning of all the parameters in the Sync Gateway documentation.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 |
público interfaz ChangesFeedListener { vacío onResponse(Cadena body); } público vacío start() { monitorThread = nuevo Thread(() -> { while (!Thread.interrupted()) { Request request = nuevo Request.Builder() .url(urlBuilder.build()) .build(); call = client.newCall(request); Intentar (Response response = call.execute()) { si (!response.isSuccessful()) lanzar nuevo IOException(“Unexpected code “ + response); Cadena body = response.body().cadena(); JsonNode tree = mapper.readTree(body); since = tree.conseguir(“last_seq”).asText(); urlBuilder.setQueryParameter(“since”, since); listener.onResponse(body); } atrapar (SocketException ex) { regresar; } atrapar (IOException ex) { ex.printStackTrace(); Dialog.display(ex); } } }); monitorThread.setDaemon(true); monitorThread.start(); } |
El start method has the most interesting part of the code. It spins up a background thread. Underneath the thread setup and control code I run a continuous loop. The loop does synchronous network calls. The error handling is simple. Just throw an exception if anything goes wrong.
Sync Gateway responds with JSON strings. You can see the code pulls apart the response and parses the JSON into a JsonNode object. This is all to get at the last_seq value in the response.
In order to track what to send next, the changes feed relies on a simple sequence mechanism. You should treat this as an opaque object. Take the value of last_seq from the previous response, and set the since parameter to that same value for the next request.
There’s no real harm in not supplying the since parameter. Sync Gateway will just replay all changes from the start if it’s missing. That’s why you’ll see in this example, I cheat a little and always create the class instance with since set to the string “0”.
In a real world application, you might want to have some way to save the last sequence string your app has processed, rather than churning through the change history every time.
The rest of the code is just a couple of short methods.
1
And that’s it for the main classes. There are others needed for the complete app.
Check out the GitHub repo to see all the code and instructions to build it.
Read a discussion of the app and how to use it in part one.
Postscript
You can find more resources on our developer portal and follow us on Twitter @CouchbaseDev.
You can post questions on our forums. And we actively participate on Stack Overflow.
Hit me up on Twitter with any questions, comments, topics you’d like to see, etc. @HodGreeley

Deja un comentario
Lo siento, debes estar conectado para publicar un comentario.