<commit_before>(ns kixi.datastore.filestore.s3
  (:require [amazonica.core :as aws]
            [amazonica.aws.s3 :as s3]
            [amazonica.aws.s3transfer :as s3t]
            [byte-streams :as bs]
            [clojure.core.async :as async :refer [go]]
            [com.stuartsierra.component :as component]
            [kixi.datastore.filestore :as fs :refer [FileStore]]
            [taoensso.timbre :as timbre :refer [error]]))

(defn ensure-bucket
  [creds bucket]
  (when-not (s3/does-bucket-exist creds bucket)
    (s3/create-bucket creds bucket)))

(def ^Integer buffer-size 
  "The piped stream must have a buffer size at least as big as the S3 clients: http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/constant-values.html#com.amazonaws.RequestClientOptions.DEFAULT_STREAM_BUFFER_SIZE"
  (inc 131073))

(defn upload
  [creds bucket id content-length]
  (let [in-stream (new java.io.PipedInputStream buffer-size)
        out-stream (new java.io.PipedOutputStream in-stream)]
    [(go
       (try
         (s3/put-object creds
                        bucket id in-stream 
                        {:content-length content-length})
         (catch Throwable t
           (error t "S3 put exception")
           (throw t))))
     out-stream]))

(defrecord S3
    [logging region endpoint access-key secret-key bucket client-options creds]
    FileStore
    (exists [this id]
      (s3/does-object-exist creds bucket id))
    (output-stream [this id content-length]
      (upload creds bucket id content-length))
    (retrieve [this id]
      (when (s3/does-object-exist creds bucket id)
        (:object-content
         (s3/get-object creds bucket id))))
    component/Lifecycle
    (start [component]      
      (if-not creds
        (let [c (merge {:endpoint endpoint}
                       (when secret-key 
                         {:secret-key secret-key})
                       (when access-key 
                         {:access-key access-key}))]
          (ensure-bucket c bucket)
          (assoc component
                 :creds
                 c))<commit_msg>Correcting the output of the go loop in S3<commit_after>(ns kixi.datastore.filestore.s3
  (:require [amazonica.core :as aws]
            [amazonica.aws.s3 :as s3]
            [amazonica.aws.s3transfer :as s3t]
            [byte-streams :as bs]
            [clojure.core.async :as async :refer [go]]
            [com.stuartsierra.component :as component]
            [kixi.datastore.filestore :as fs :refer [FileStore]]
            [taoensso.timbre :as timbre :refer [error]]))

(defn ensure-bucket
  [creds bucket]
  (when-not (s3/does-bucket-exist creds bucket)
    (s3/create-bucket creds bucket)))

(def ^Integer buffer-size 
  "The piped stream must have a buffer size at least as big as the S3 clients: http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/constant-values.html#com.amazonaws.RequestClientOptions.DEFAULT_STREAM_BUFFER_SIZE"
  (inc 131073))

(defn upload
  [creds bucket id content-length]
  (let [in-stream (new java.io.PipedInputStream buffer-size)
        out-stream (new java.io.PipedOutputStream in-stream)]
    [(go
       (try
         (s3/put-object creds
                        bucket id in-stream 
                        {:content-length content-length})
         :done
         (catch Throwable t
           t)))
     out-stream]))

(defrecord S3
    [logging region endpoint access-key secret-key bucket client-options creds]
    FileStore
    (exists [this id]
      (s3/does-object-exist creds bucket id))
    (output-stream [this id content-length]
      (upload creds bucket id content-length))
    (retrieve [this id]
      (when (s3/does-object-exist creds bucket id)
        (:object-content
         (s3/get-object creds bucket id))))
    component/Lifecycle
    (start [component]      
      (if-not creds
        (let [c (merge {:endpoint endpoint}
                       (when secret-key 
                         {:secret-key secret-key})
                       (when access-key 
                         {:access-key access-key}))]
          (ensure-bucket c bucket)
          (assoc component
                 :creds
                 c))
{'fraction_non_alphanumeric': 0.10656130268199233, 'fraction_numerical': 0.009578544061302681, 'mean_word_length': 1.9921203438395416, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4608196', 'n_tokens_mistral': 1303, 'n_tokens_neox': 1265, 'n_words': 334}