       (assoc this
              :segment segment
              :rst remaining
-             :offset (if segment (inc offset) offset)
-             :completed? (empty? remaining))))
+             :offset (if segment (inc offset) offset))))
 
-  (segment-complete! [{:keys [conn]} segment])
+  (segment-complete! [this segment])
 
   (completed? [this]
-    (:completed? this)))
+    (empty? rst)))
 
 (defn input [event]
   (map->AbsSeqReader {:event event}))

            [onyx.plugin.onyx-output :as o]
            [onyx.plugin.onyx-plugin :as p]))

(defrecord AbsSeqReader [event completed? sequential rst segment offset]
  p/OnyxPlugin

  (start [this]

    offset)

  (recover [this checkpoint]
    (if (nil? checkpoint) 
      (assoc this 
             :rst sequential 
             :offset -1)
      (assoc this 
             :rst (drop (inc checkpoint) sequential)
             :offset checkpoint)))

  (offset-id [this]
    offset)
