The Couchbase Mobile<\/a> Sync Gateway<\/a> 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<\/a> of this two part series. The code also serves as an example of listening to the feed.<\/p>\n 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<\/a> for updates. \u00a0You can also find instructions for building, running, and packaging the app there.<\/p>\n JavaFX breaks simple apps into a controller class and a declarative UI. Let’s walk through the controller.<\/p>\n 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.<\/p>\n The Next comes initialization. JavaFX calls this method as part of its standard lifecycle.<\/p>\n I’ve broken out the document list initialization into its own routine. The document list gets bound to the 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 With the other bindings in place, we just have to update the The next few lines set the initial state of a couple of toggle buttons. I need an extra listener to keep the I wrote a separate class to monitor Sync Gateway. The initialization code finished by creating a new monitor instance and kicking it off.<\/p>\n The next section contains several listeners.<\/p>\n 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 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.<\/p>\n I used a callback approach to get the results of the changes feed. The interface is defined in the 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 here<\/a>. For this app I just need to know whether a replication is running or not to keep the Next we have several methods bound to UI elements. JavaFX handles much of the wiring.<\/p>\n Here I set the use of authentication credentials whenever the corresponding button gets toggled.<\/p>\n This code shows a couple of interesting items. I use a Jackson Next I check for an entry 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 here<\/a>.<\/p>\n This reacts to toggling the The rest of the code here are just helper bits and a piece to shutdown everything before exiting.<\/p>\n 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.<\/p>\n I do implement the Finally, let’s take a look at the helper class for monitoring Sync Gateway. I’ll walk through this in pieces, too.<\/p>\n I use the OkHttp library from Square<\/a>. 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<\/a>.<\/p>\n The Sync Gateway responds with JSON strings. You can see the code pulls apart the response and parses the JSON into a 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 There’s no real harm in not supplying the 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.<\/p>\n The rest of the code is just a couple of short methods.<\/p>\n And that’s it for the main classes. There are others needed for the complete app.<\/p>\n Check out the GitHub repo<\/a> to see all the code and instructions to build it.<\/p>\n Read a discussion of the app and how to use it in part one<\/a>.<\/p>\n You can find more resources on our developer portal<\/a> and follow us on Twitter @CouchbaseDev<\/a>.<\/p>\n You can post questions on our forums<\/a>. And we actively participate on Stack Overflow<\/a>.<\/p>\nThe code<\/h2>\n
JavaFX: The Controller class<\/h3>\n
package com.couchbase.mobile;\r\n\r\nimport com.couchbase.lite.*;\r\nimport com.fasterxml.jackson.core.JsonProcessingException;\r\nimport javafx.application.Platform;\r\nimport javafx.beans.value.ChangeListener;\r\nimport javafx.beans.value.ObservableValue;\r\nimport javafx.collections.FXCollections;\r\nimport javafx.collections.ObservableList;\r\nimport javafx.event.ActionEvent;\r\nimport javafx.fxml.FXML;\r\nimport javafx.scene.control.*;\r\nimport javafx.scene.control.TextField;\r\n\r\nimport java.io.IOException;\r\nimport java.net.MalformedURLException;\r\nimport java.net.URL;\r\nimport java.util.Map;\r\n\r\nimport static com.couchbase.mobile.Runtime.mapper;\r\n\r\npublic class Controller implements LiveQuery.ChangeListener, ChangeListener,\r\n SGMonitor.ChangesFeedListener, DBService.ReplicationStateListener {\r\n private static final String SYNC_GATEWAY_HOST = \"https:\/\/localhost\";\r\n private static final String SG_PUBLIC_URL = SYNC_GATEWAY_HOST + \":4984\/\" + DBService.DATABASE;\r\n private static final String SG_ADMIN_URL = SYNC_GATEWAY_HOST + \":4985\/\" + DBService.DATABASE;\r\n\r\n private static final String TOGGLE_INACTIVE = \"-fx-background-color: #e6555d;\";\r\n private static final String TOGGLE_ACTIVE = \"-fx-background-color: #ade6a6;\";\r\n private static final String TOGGLE_DISABLED = \"-fx-background-color: #555555;\";\r\n\r\n @FXML private ListView documentList;\r\n private ObservableList documents = FXCollections.observableArrayList();\r\n @FXML private TextArea contentsText;\r\n @FXML private TextArea changesFeed;\r\n @FXML private TextField usernameText;\r\n @FXML private TextField passwordText;\r\n @FXML private ToggleButton applyCredentialsBtn;\r\n @FXML private ToggleButton syncBtn;\r\n\r\n private DBService service = DBService.getInstance();\r\n private Database db = service.getDatabase();\r\n private SGMonitor changesMonitor;\r\n private LiveQuery liveQuery;\r\n<\/pre>\n@FXML<\/code> annotations mark all the fields that the framework will automatically bind to portions of the UI.<\/p>\n @FXML private void initialize() {\r\n documentListInitialize();\r\n documentList.setItems(documents);\r\n\r\n setState(applyCredentialsBtn, false);\r\n setState(syncBtn, false);\r\n\r\n service.addReplicationStateListener(this);\r\n\r\n changesMonitor = new SGMonitor(SG_ADMIN_URL, \"false\", \"true\", \"0\", \"all_docs\", this);\r\n changesMonitor.start();\r\n }\r\n\r\n private void documentListInitialize() {\r\n Query query = db.createAllDocumentsQuery();\r\n query.setAllDocsMode(Query.AllDocsMode.INCLUDE_DELETED);\r\n liveQuery = query.toLiveQuery();\r\n liveQuery.addChangeListener(this);\r\n liveQuery.start();\r\n\r\n documentList.getSelectionModel().selectedItemProperty().addListener(this);\r\n }<\/pre>\ndocumentList<\/code> variable. In turn documentList<\/code> will update the UI whenever the item list we pass in changes.<\/p>\nINCLUDE_DELETED<\/code> so the tool can show what a deleted document looks like in the database.<\/p>\ndocuments<\/code> list. We’ll see the live query listener that does that further along.<\/p>\nSync<\/code> button consistent with the actual state of the replications. More on this further along in the article.<\/p>\n \/\/ LiveQuery.ChangeListener\r\n @Override\r\n public void changed(LiveQuery.ChangeEvent event) {\r\n if (event.getSource().equals(liveQuery)) {\r\n Platform.runLater(() -> {\r\n QueryEnumerator rows = event.getRows();\r\n\r\n documents.clear();\r\n\r\n rows.forEach(queryRow -> documents.add(queryRow.getDocumentId()));\r\n });\r\n }\r\n }<\/pre>\ngetRows<\/code> method returns an enumerator that will index doing just that. JavaFX takes care of updating the UI when documents<\/code> changes.<\/p>\n \/\/ ListView ChangeListener\r\n @Override\r\n public void changed(ObservableValue observable, String oldId, String newId) {\r\n if (null == newId) return;\r\n\r\n Map properties = db.getDocument(newId).getProperties();\r\n\r\n try {\r\n String json = mapper.writeValueAsString(properties);\r\n\r\n contentsText.setText(prettyText(json));\r\n } catch (JsonProcessingException ex) {\r\n ex.printStackTrace();\r\n Dialog.display(ex);\r\n }\r\n }<\/pre>\n \/\/ SGMonitor.ChangesFeedListener\r\n @Override\r\n public void onResponse(String body) {\r\n changesFeed.appendText(prettyText((String) body));\r\n }<\/pre>\nSGMonitor<\/code> 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.<\/p>\n \/\/ DBService.ReplicationStateListener\r\n @Override\r\n public void onChange(boolean isActive) {\r\n setState(syncBtn, isActive);\r\n }<\/pre>\nSync<\/code> 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.<\/p>\n @FXML private void applyCredentialsToggled(ActionEvent event) {\r\n String username = null;\r\n String password = null;\r\n\r\n if (applyCredentialsBtn.isSelected()) {\r\n username = usernameText.getText();\r\n password = passwordText.getText();\r\n }\r\n\r\n DBService.getInstance().setCredentials(username, password);\r\n applyCredentialsBtn.setStyle(applyCredentialsBtn.isSelected() ? TOGGLE_ACTIVE : TOGGLE_INACTIVE);\r\n }<\/pre>\n @FXML private void saveContentsClicked(ActionEvent event) {\r\n Map properties = null;\r\n Document document;\r\n\r\n try {\r\n properties = mapper.readValue(contentsText.getText(), Map.class);\r\n } catch (IOException ex) {\r\n ex.printStackTrace();\r\n Dialog.display(ex);\r\n }\r\n\r\n if (properties.containsKey(\"_id\")) {\r\n document = db.getDocument((String) properties.get(\"_id\"));\r\n } else {\r\n document = db.createDocument();\r\n }\r\n\r\n try {\r\n document.putProperties(properties);\r\n } catch (CouchbaseLiteException ex) {\r\n ex.printStackTrace();\r\n Dialog.display(ex);\r\n }\r\n }\r\n<\/pre>\nObjectMapper<\/code> instance to convert the text in the content pane to a property map.<\/p>\n_id<\/code>. Couchbase Mobile reserves most properties starting with an “_” for system use (with special exceptions). If the text we’re trying to convert contains _id<\/code>, I assume this is an edit to an existing document. Otherwise I create a new document.<\/p>\n @FXML private void syncToggled(ActionEvent event) {\r\n try {\r\n syncBtn.setDisable(true);\r\n syncBtn.setStyle(TOGGLE_DISABLED);\r\n service.toggleReplication(new URL(SG_PUBLIC_URL), true);\r\n } catch (Exception ex) {\r\n ex.printStackTrace();\r\n Dialog.display(ex);\r\n syncBtn.setDisable(false);\r\n }\r\n }<\/code><\/pre>\nSync<\/code> button. Recall though that we use a listener to verify the state elsewhere.<\/p>\n @FXML private void exitClicked(ActionEvent event) {\r\n \/\/ Try to shut everything down gracefully\r\n changesMonitor.stop();\r\n liveQuery.stop();\r\n service.stopReplication();\r\n db.close();\r\n db.getManager().close();\r\n\r\n Platform.exit();\r\n }\r\n\r\n private void setState(ToggleButton btn, boolean active) {\r\n btn.setSelected(active);\r\n btn.setStyle(active ? TOGGLE_ACTIVE : TOGGLE_INACTIVE);\r\n btn.setDisable(false);\r\n }\r\n\r\n private String prettyText(String json) {\r\n String out = null;\r\n\r\n try {\r\n Object object = mapper.readValue(json, Object.class);\r\n\r\n out = mapper.writerWithDefaultPrettyPrinter().writeValueAsString(object);\r\n } catch (Exception ex) {\r\n ex.printStackTrace();\r\n }\r\n\r\n return out;\r\n }\r\n}\r\n<\/pre>\nThe Database Helper class<\/h3>\n
Replication.ChangeListener<\/code> interface. That’s maybe a little unusual. I mentioned the reason earlier on. This link takes you to the blog post<\/a> about it.<\/p>\npackage com.couchbase.mobile;\r\n\r\nimport com.couchbase.lite.Database;\r\nimport com.couchbase.lite.JavaContext;\r\nimport com.couchbase.lite.Manager;\r\nimport com.couchbase.lite.auth.Authenticator;\r\nimport com.couchbase.lite.auth.AuthenticatorFactory;\r\nimport com.couchbase.lite.replicator.Replication;\r\nimport com.couchbase.lite.replicator.ReplicationState;\r\n\r\nimport java.net.URL;\r\nimport java.util.ArrayList;\r\nimport java.util.List;\r\n\r\npublic class DBService implements Replication.ChangeListener {\r\n public static final String DATABASE = \"db\";\r\n\r\n private static final String DB_DIRECTORY = \"data\";\r\n\r\n private Manager manager;\r\n private Database database;\r\n private Replication pushReplication = null;\r\n private Replication pullReplication = null;\r\n private boolean replicationActive = false;\r\n private List stateListeners = new ArrayList();\r\n private String username = null;\r\n private String password = null;\r\n\r\n private DBService() {\r\n try {\r\n manager = new Manager(new JavaContext(DB_DIRECTORY), Manager.DEFAULT_OPTIONS);\r\n database = manager.getDatabase(DATABASE);\r\n } catch (Exception ex) {\r\n ex.printStackTrace();\r\n }\r\n }\r\n\r\n private static class Holder {\r\n private static DBService INSTANCE = new DBService();\r\n }\r\n\r\n\r\n public interface ReplicationStateListener {\r\n void onChange(boolean isActive);\r\n }\r\n\r\n public static DBService getInstance() {\r\n return Holder.INSTANCE;\r\n }\r\n\r\n public Database getDatabase() {\r\n return database;\r\n }\r\n\r\n public void setCredentials(String username, String password) {\r\n this.username = username;\r\n this.password = password;\r\n }\r\n\r\n public void toggleReplication(URL gateway, boolean continuous) {\r\n if (replicationActive) {\r\n stopReplication();\r\n } else {\r\n startReplication(gateway, continuous);\r\n }\r\n }\r\n\r\n public void startReplication(URL gateway, boolean continuous) {\r\n if (replicationActive) {\r\n stopReplication();\r\n }\r\n\r\n pushReplication = database.createPushReplication(gateway);\r\n pullReplication = database.createPullReplication(gateway);\r\n pushReplication.setContinuous(continuous);\r\n pullReplication.setContinuous(continuous);\r\n\r\n if (username != null) {\r\n Authenticator auth = AuthenticatorFactory.createBasicAuthenticator(username, password);\r\n pushReplication.setAuthenticator(auth);\r\n pullReplication.setAuthenticator(auth);\r\n }\r\n\r\n pushReplication.addChangeListener(this);\r\n pullReplication.addChangeListener(this);\r\n\r\n pushReplication.start();\r\n pullReplication.start();\r\n }\r\n\r\n public void stopReplication() {\r\n if (!replicationActive) return;\r\n\r\n pushReplication.stop();\r\n pullReplication.stop();\r\n\r\n pushReplication = null;\r\n pullReplication = null;\r\n }\r\n\r\n public void addReplicationStateListener(ReplicationStateListener listener) {\r\n stateListeners.add(listener);\r\n }\r\n\r\n public void removeReplicationStateListener(ReplicationStateListener listener) {\r\n stateListeners.remove(listener);\r\n }\r\n\r\n \/\/ Replication.ChangeListener\r\n @Override\r\n public void changed(Replication.ChangeEvent changeEvent) {\r\n if (changeEvent.getError() != null) {\r\n Throwable lastError = changeEvent.getError();\r\n\r\n Dialog.display(lastError.getMessage());\r\n\r\n return;\r\n }\r\n\r\n if (changeEvent.getTransition() == null) return;\r\n\r\n ReplicationState dest = changeEvent.getTransition().getDestination();\r\n\r\n replicationActive = ((dest == ReplicationState.STOPPING || dest == ReplicationState.STOPPED) ? false : true);\r\n\r\n stateListeners.forEach(listener -> listener.onChange(replicationActive));\r\n }\r\n}<\/pre>\nThe Sync Gateway Monitor class<\/h3>\n
package com.couchbase.mobile;\r\n\r\nimport com.fasterxml.jackson.databind.JsonNode;\r\nimport okhttp3.*;\r\n\r\nimport java.io.IOException;\r\nimport java.net.SocketException;\r\nimport java.util.concurrent.TimeUnit;\r\n\r\nimport static com.couchbase.mobile.Runtime.mapper;\r\n\r\npublic class SGMonitor {\r\n private static final OkHttpClient client = new OkHttpClient.Builder()\r\n .readTimeout(1, TimeUnit.DAYS)\r\n .build();\r\n private ChangesFeedListener listener;\r\n private HttpUrl.Builder urlBuilder;\r\n private Thread monitorThread;\r\n private String since = \"0\";\r\n private Call call;\r\n\r\n SGMonitor(String url, String activeOnly, String includeDocs, String since, String style,\r\n ChangesFeedListener listener) {\r\n this.since = since;\r\n\r\n urlBuilder = HttpUrl.parse(url).newBuilder()\r\n .addPathSegment(\"_changes\")\r\n .addQueryParameter(\"active_only\", activeOnly)\r\n .addQueryParameter(\"include_docs\", includeDocs)\r\n .addQueryParameter(\"style\", style)\r\n .addQueryParameter(\"since\", since)\r\n .addQueryParameter(\"feed\", \"longpoll\")\r\n .addQueryParameter(\"timeout\", \"0\");\r\n\r\n this.listener = listener;\r\n }<\/pre>\n public interface ChangesFeedListener {\r\n void onResponse(String body);\r\n }\r\n\r\n public void start() {\r\n monitorThread = new Thread(() -> {\r\n while (!Thread.interrupted()) {\r\n Request request = new Request.Builder()\r\n .url(urlBuilder.build())\r\n .build();\r\n\r\n call = client.newCall(request);\r\n\r\n try (Response response = call.execute()) {\r\n if (!response.isSuccessful()) throw new IOException(\"Unexpected code \" + response);\r\n\r\n String body = response.body().string();\r\n\r\n JsonNode tree = mapper.readTree(body);\r\n\r\n since = tree.get(\"last_seq\").asText();\r\n urlBuilder.setQueryParameter(\"since\", since);\r\n\r\n listener.onResponse(body);\r\n } catch (SocketException ex) {\r\n return;\r\n } catch (IOException ex) {\r\n ex.printStackTrace();\r\n Dialog.display(ex);\r\n }\r\n }\r\n });\r\n\r\n monitorThread.setDaemon(true);\r\n monitorThread.start();\r\n }<\/code><\/pre>\nstart<\/code> 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.<\/p>\nJsonNode<\/code> object. This is all to get at the last_seq<\/code> value in the response.<\/p>\nlast_seq<\/code> from the previous response, and set the since<\/code> parameter to that same value for the next request.<\/p>\nsince<\/code> 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<\/code> set to the string “0”.<\/p>\n public void stop() {\r\n monitorThread.interrupt();\r\n call.cancel();\r\n }\r\n\r\n public String getSince() {\r\n return since;\r\n }\r\n}<\/pre>\nPostscript<\/h2>\n