Description of Question 1 solution.

question1.clj

;; Hardcode the input data here as `data`

(def data `({:content "Message from User 2"
             :from {:id 2 :name "User 2" :type :user :updated "2000-03-05T16:49:56.845Z"}
             :to {:id 22 :name "Group 22" :type :group :updated "1991-10-19T00:57:39.527Z"}
             :type :message}

            {:content "Message from User 1"
             :from {:id 1 :name "User 1" :type :user :updated "1980-07-21T18:56:27.379Z"}
             :to {:id 3 :name "User 3" :type :user :updated "1984-05-16T19:57:44.739Z"}
             :type :message}

            {:content "Message from User 0"
             :from {:id 0 :name "User 0" :type :user :updated "2002-02-07T11:53:37.041Z"}
             :to {:id 4 :name "User 4" :type :user :updated "1997-05-04T14:09:15.871Z"}
             :type :message}

            {:content "Message from User 2"
             :from {:id 2 :name "User 2" :type :user :updated "1999-08-09T21:24:37.760Z"}
             :to {:id 25 :name "Group 25" :type :group :updated "2005-06-24T07:18:02.670Z"}
             :type :message}

            {:content "Message from User 1"
             :from {:id 1 :name "User 1" :type :user :updated "1983-11-04T09:39:02.995Z"}
             :to {:id 4 :name "User 4" :type :user :updated "1980-02-19T14:00:17.164Z"}
             :type :message}))

;; The output is just for reference

(def output `({:id 2 :name "User 2" :type :user :updated "2000-03-05T16:49:56.845Z"}
              {:id 22 :name "Group 22" :type :group :updated "1991-10-19T00:57:39.527Z"}
              {:content "Message from User 2" :from 2 :to 22 :type :message}
              {:id 1 :name "User 1" :type :user :updated "1980-07-21T18:56:27.379Z"}
              {:id 3 :name "User 3" :type :user :updated "1984-05-16T19:57:44.739Z"}
              {:content "Message from User 1" :from 1 :to 3 :type :message}
              {:id 0 :name "User 0" :type :user :updated "2002-02-07T11:53:37.041Z"}
              {:id 4 :name "User 4" :type :user :updated "1997-05-04T14:09:15.871Z"}
              {:content "Message from User 0" :from 0 :to 4 :type :message}
              {:id 25 :name "Group 25" :type :group :updated "2005-06-24T07:18:02.670Z"}
              {:content "Message from User 2" :from 2 :to 25 :type :message}
              {:id 1 :name "User 1" :type :user :updated "1983-11-04T09:39:02.995Z"}
              {:content "Message from User 1" :from 1 :to 4 :type :message}))


;; Setup few helper functions to make life easier -- all private functions

(def sdf (java.text.SimpleDateFormat. "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"))
(defn- to-time
  [updated-date]
  (.getTime (.parse sdf updated-date)))

(defn- is-cached?
  [cache id]
  (.containsKey cache id))

(defn- parse-user-or-group-data
  "Check whether the id is in the MRU cache,
   compare and returns the most recent record if it's in it,
   else cache and returns the data"
  [cache msg]
  (let [id (:id msg)
        updated (to-time (:updated msg))]
        (if (is-cached? cache id)
          (if (< (.get cache id) updated)
            msg
            ())
          (do
            (.put cache id updated)
            msg))))

(defn- parse-message
  "Parse the stream data into its component structs, conforming to the 'rules'"
  [cache msg]
  (list (parse-user-or-group-data cache (msg :from))
        (parse-user-or-group-data cache (msg :to))
        (assoc (dissoc msg :from :to) :from (get-in msg [:from :id]) :to (get-in msg [:to :id]))))


;; Note the use of com.ping.MRUCache class, a simple implementation of MRU cache using LinkedHashMap.

(defn transform-stream
  [cache-size stream]
  (let [cache (com.ping.MRUCache. cache-size)
        result (-> (for [x stream] (parse-message cache x)) flatten)] result) )


;; Helper function to kickstart the whole thing.

(defn run []
  (transform-stream (* 3 (count data)) data))



;; //////////////////////////////////////////////////////////////////////////////////////////////////////////////
;; Naive MRU Cache implemented in Java.
;; MRUCache.java
;; //////////////////////////////////////////////////////////////////////////////////////////////////////////////
package com.ping;

import java.util.LinkedHashMap;
import java.util.Map;


public class MRUCache extends LinkedHashMap<Integer, Long> {
  private static float loadFactor = 0.75f;
  private int cacheSize;

  public MRUCache(int cacheSize) {
    super(cacheSize, loadFactor, false);
    this.cacheSize = cacheSize;
  }

  @Override
  protected boolean removeEldestEntry(Map.Entry<Integer, Long> eldest) {
    return size() >= cacheSize;
  }

}

Here is the result output from repl.

➜  assignment lein repl
Compiling 1 source files to /Users/ping/Work/assignment/target/base+system+user+dev/classes
nREPL server started on port 60314 on host 127.0.0.1 - nrepl://127.0.0.1:60314
REPL-y 0.3.7, nREPL 0.2.12
Clojure 1.8.0
Java HotSpot(TM) 64-Bit Server VM 1.8.0_60-b27
    Docs: (doc function-name-here)
          (find-doc "part-of-name-here")
  Source: (source function-name-here)
 Javadoc: (javadoc java-object-or-class-here)
    Exit: Control+D or (exit) or (quit)
 Results: Stored in vars *1, *2, *3, an exception in *e

assignment.core=> (run)
({:id 2 :name "User 2" :type :user :updated "2000-03-05T16:49:56.845Z"}
 {:id 22 :name "Group 22" :type :group :updated "1991-10-19T00:57:39.527Z"}
 {:content "Message from User 2" :from 2 :to 22 :type :message}
 {:id 1 :name "User 1" :type :user :updated "1980-07-21T18:56:27.379Z"}
 {:id 3 :name "User 3" :type :user :updated "1984-05-16T19:57:44.739Z"}
 {:content "Message from User 1" :from 1 :to 3 :type :message}
 {:id 0 :name "User 0" :type :user :updated "2002-02-07T11:53:37.041Z"}
 {:id 4 :name "User 4" :type :user :updated "1997-05-04T14:09:15.871Z"}
 {:content "Message from User 0" :from 0 :to 4 :type :message}
 {:id 25 :name "Group 25" :type :group :updated "2005-06-24T07:18:02.670Z"}
 {:content "Message from User 2" :from 2 :to 25 :type :message}
 {:id 1 :name "User 1" :type :user :updated "1983-11-04T09:39:02.995Z"}
 {:content "Message from User 1" :from 1 :to 4 :type :message})
assignment.core=>