Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Persistent Queues

Persistent Queues

Offloading tasks to the background is a great way to simulate the feeling of responsiveness. But a lot can go wrong. This talk takes a deep dive into the implementation of Tape, a library that lets your app reliably process tasks in the background. It explores the API, suggests patterns for common use cases, and suggest alternatives for when Tape's simplicity isn't enough.

Prateek Srivastava

October 27, 2015
Tweet

More Decks by Prateek Srivastava

Other Decks in Programming

Transcript

  1. “rwd” Open for reading and writing, as with "rw", and

    also require that every update to the file's content be written synchronously to the underlying storage device.
  2. = =

  3. =

  4. API

  5. class QueueFile { void add(byte[] data); byte[] peek(); void remove();

    void clear(); int size(); void forEach(ElementReader reader); } @1.2.3
  6. // Read the data at the head of the queue.

    byte[] element = queueFile.peek(); // Process it. process(element); // Remove if from the queue. queueFile.remove();
  7. example byte[] message = queueFile.peek(); if (message != null) {

    return; } HttpURLConnection connection = open("https://segment.com/import"); connection.getOutputStream().write(message);
  8. try { connection.close(); } catch (IOException e) { // Network

    error, retry later. return; } int responseCode = connection.getResponseCode(); if (responseCode >= 500) { // Server error, retry later. return; } // Client error, e.g. invalid json. Don’t retry. if (responseCode >= 400) { log("server rejected message"); } queueFile.remove(); example
  9. example if (queueFile.size() < 0) { return; } HttpURLConnection connection

    = open("https://segment.com/import"); OutputStream out = connection.getOutputStream(); queueFile.forEach((in, length) -> { Utils.copy(in, out); });
  10. example for (int i = 0; i < n; i++)

    { queueFile.remove(); }
  11. example HttpURLConnection connection = open("https://segment.com/import"); OutputStream out = connection.getOutputStream(); *int

    count = 0; // simplified queueFile.forEach((in, length) -> { if (count + length > MAX_LIMIT) { return; } count += Utils.copy(in, out); return; });
  12. example HttpURLConnection connection = open("https://segment.com/import"); OutputStream out = connection.getOutputStream(); *int

    count = 0; // simplified queueFile.forEach((in, length) -> { if (count + length > MAX_LIMIT) { return false; } count += Utils.copy(in, out); return true; });
  13. example HttpURLConnection connection = open("https://segment.com/import"); OutputStream out = connection.getOutputStream(); *int

    count = 0; // simplified queueFile.forEach((in, length) -> { if (count + length > MAX_LIMIT) { return false; } count += Utils.copy(in, out); return true; });
  14. @1.2.3 interface ObjectQueue<T> { void add(T entry); T peek(); void

    remove(); int size(); void setListener(Listener<T> listener); }