   function takes three parameters: the topic, the size and whether to
   receive the cache items as a response or as published messsages. The
   cache data is availabe in the source state record."
-  [{:keys [cache] :as source} topic size publish?]
+  [{:keys [topics] :as source} topic size publish?]
   (debug "Requested to send " size "items from cache for" topic "to" wamp/*call-sess-id*)
-  (let [messages (reverse (take size (get @cache topic)))
+  (let [messages (reverse (take size (:cache @(get @topics topic))))
         cnt (count messages)]
     (if publish?
       (do (debug "Sending" cnt "cached message for" topic "to" wamp/*call-sess-id* "as events.")


;;; Source implementation.

;; topics = (ref #{"topic"})
;; cache = (atom {"topic" (msg-3 msg-2 msg-1)})
;; open = (atom boolean)
;; config = {:cache-size long, :clean-cache-on-done boolean, :clean-summary-on-done boolean}
;; summaries (atom {"topic", summary})
(defrecord WebsocketSource [dirwriter stop-fn open topics cache uri config summary-fn summaries]
  Source
  (publish [this topic message]
    (let [uuid (UUID/randomUUID)]
      (debug "Requested to publishing message" uuid "on" topic)
      (trace "Message" uuid "-" message)
      (when @open
        (when (dosync
                (when-not (get @topics topic)
                  (alter topics conj topic)))
          (when @open
            (api/add-source dirwriter topic uri)
            (debug "Registered" uri "for topic" topic "in directory service.")))
        (wamp/send-event! topic message)
        (let [new-cache (swap! cache update-in [topic]
                               (fn [c] (take (config :cache-size) (conj c message))))]
          (debug "Message " uuid "published and cached.")
          (trace "Cache for" topic "after message" uuid "is:" (get new-cache topic)))
        (when summary-fn
          (let [new-summaries (swap! summaries update-in [topic] summary-fn message)]
            (debug "Summary for" topic "updated with message" uuid)
            (trace "Summary for" topic "after message" uuid "is:" (get new-summaries topic)))))))

  (done [this topic]
    (debug "Initiated 'done' for" topic)
    (when @open
      (when (dosync (when (get @topics topic)
                      (alter topics disj topic)))
        (when @open
          (when-let [done-payload (:done-payload config)]
            (debug "Sending done payload to" topic)
            (wamp/send-event! topic done-payload))
          (api/remove-source dirwriter topic uri)
          (debug "Unregistered" uri "for topic" topic "in directory service."))
        (when (:clean-cache-on-done config)
          (swap! cache dissoc topic)
          (debug "Cleaned cache for" topic))
        (when (and summary-fn (:clean-summary-on-done config))
          (swap! summaries dissoc topic)
          (debug "Cleaned summary for" topic))))))


(defn start-source

  "Given the return value of `start-source`, this stops the Websocket
  server and removes every topic for this source from the directory
  service."
  [{:keys [stop-fn open dirwriter topics config] :as source}]
  (when @open
    (info "Stopping websocket source, using state" source)
    (reset! open false)
    (doseq [topic-agent (vals @topics)]
      (send topic-agent done-action config))
    (@stop-fn :timeout 100)))
