clj-gcp

Clojure utilities for the Google Cloud Platform.

Usage

Pub-Sub

Mainly used through integrant:

;; ig-config.edn
:clj-gcp.pub-sub.core/subscriber
{:metrics-registry     #ig/ref :metrics/registry ;; a iapetos metric registry
 :handler              #ig/ref :pubsub/handler ;; a fn, described below
 :project-id                   "my-gcp-project"
 :topic-id                     "industry-data-lake-file-notifications"
 :pull-max-messages            10
 :subscription-id              "LOCAL_DEV.bucket-notifications.my-service"}

The function :handler takes a seq of maps that contain:

{,,, pub-sub message fields (always deserialized as JSON) ,,,
 :pubsub/attributes {:eventType "OBJECT_FINALIZE"}
 :pubsub/ack-id     "0000000ACK"}

... and should return a seq of maps that contain at least:

;; these will be acked:
{:pubsub/ack-id     "0000000ACK"
 :ok? true}
{:pubsub/ack-id     "0000000ACK"
 :ok? false
 :retry? false}

;; this will be nacked:
{:pubsub/ack-id     "0000000ACK"
 :ok? false
 :retry? true}

Healthcheck

There's also a healthcheck integrant key available:

;; ig-config.edn
:clj-gcp.pub-sub.core/subscriber.healthcheck
{:project-id     "my-gcp-project"
 :subscription-id "LOCAL_DEV.bucket-notifications.my-service"}

Storage

(let [sut                 (sut/->gcs-storage-client)
      blob-name           (format "TEST_BLOBS/BLOB-%s.txt" (m/random-uuid))
      exp-contentEncoding "utf-8"
      exp-content         "hello!"
      exp-metadata        {"hi" "foo"}
      exp-contentType     "text/plain"]

  ;; This is how you write a blob:
  (with-open [out (Channels/newWriter
                   (sut/blob-writer sut
                                    bucket-name
                                    blob-name
                                    {:metadata        exp-metadata,
                                     :contentType     exp-contentType,
                                     :contentEncoding exp-contentEncoding})
                   exp-contentEncoding)]
    (.write out exp-content))

  ;; ... and this is how you get it back
  (let [{:keys [inputStream md5Hash contentType source
                contentEncoding metadata],
         :as   got}
        (sut/get-blob sut bucket-name blob-name)]
    (tu/is-valid ::sut/blob got)
    (is (= (str "gs://" bucket-name "/" blob-name) source))
    (is (= exp-contentType contentType))
    (is (= exp-contentEncoding contentEncoding))
    (is (= exp-metadata metadata))
    (is (= exp-content (slurp inputStream)))))

Can also be used as an Integrant component:

;; ig-config.edn
:common.clients.storage/client             nil
:common.clients.storage/client.healthcheck nil

Testing implementation

A test implementation based on the file system is available:

(sut/->file-system-storage-client base-path)

Running Tests

GCP_PROJECT_ID=my-gcp-project lein test :integration

License

Copyright © 2018 OVO Energy Ltd.

Distributed under the Eclipse Public License either version 1.0 or (at your option) any later version.