Initial commit. kernel_info_request and kernel_info_reply implemented.

IHaskell.cabal

+-- Initial IHaskell.cabal generated by cabal init.  For further 
+-- documentation, see http://haskell.org/cabal/users-guide/
+
+-- The name of the package.
+name:                IHaskell
+
+-- The package version.  See the Haskell package versioning policy (PVP) 
+-- for standards guiding when and how versions should be incremented.
+-- http://www.haskell.org/haskellwiki/Package_versioning_policy
+-- PVP summary:      +-+------- breaking API changes
+--                   | | +----- non-breaking API additions
+--                   | | | +--- code changes with no API change
+version:             0.1.0.0
+
+-- A short (one-line) description of the package.
+synopsis:            A Haskell backend kernel for the IPython project.
+
+-- A longer description of the package.
+-- description:         
+
+-- URL for the project homepage or repository.
+homepage:            http://www.github.com/gibiansky/IHaskell
+
+-- The license under which the package is released.
+license:             GPL-3
+
+-- The file containing the license text.
+license-file:        LICENSE
+
+-- The package author(s).
+author:              Andrew Gibiansky
+
+-- An email address to which users can send suggestions, bug reports, and 
+-- patches.
+maintainer:          andrew.gibiansky@gmail.com
+
+-- A copyright notice.
+-- copyright:           
+
+category:            Development
+
+build-type:          Simple
+
+-- Constraint on the version of Cabal needed to build this package.
+cabal-version:       >=1.8
+
+
+executable IHaskell
+  -- .hs or .lhs file containing the Main module.
+  main-is:             Main.hs
+  
+  -- Modules included in this executable, other than Main.
+  -- other-modules:       
+  extensions: DoAndIfThenElse
+              NoImplicitPrelude
+              OverloadedStrings
+             
+  
+  -- Other library packages from which modules are imported.
+  build-depends:       base ==4.6.*,
+                       zeromq-haskell ==0.8.*,
+                       aeson ==0.6.*,
+                       MissingH ==1.2.*,
+                       basic-prelude ==0.3.*,
+                       unix ==2.6.*,
+                       bytestring ==0.10.*,
+                       transformers ==0.3.*,
+                       uuid ==1.2.*,
+                       containers ==0.5.*
+  

IHaskell/MessageParser.hs

+module IHaskell.MessageParser (parseMessage) where
+
+import BasicPrelude
+import Data.Aeson
+import Data.Aeson.Types (parse)
+
+import qualified Data.ByteString.Lazy as Lazy
+
+import IHaskell.Types
+
+parseHeader :: [ByteString] -> ByteString -> ByteString -> ByteString -> MessageHeader
+parseHeader idents headerData parentHeader metadata = MessageHeader {
+  identifiers = idents,
+  parentHeader = parentResult,
+  metadata = metadataMap,
+  messageId = messageUUID,
+  sessionId = sessionUUID,
+  username = username,
+  msgType = messageType
+  } where
+      -- Decode the header data and the parent header data into JSON objects.
+      -- If the parent header data is absent, just have Nothing instead.
+      Just result = decode $ Lazy.fromStrict headerData :: Maybe Object
+      parentResult = if parentHeader == "{}"
+                     then Nothing
+                     else Just $ parseHeader idents parentHeader "{}" metadata
+
+      -- Get the basic fields from the header.
+      Success (messageType, username, messageUUID, sessionUUID) = flip parse result $ \obj -> do 
+        messType <- obj .: "msg_type"
+        username <- obj .: "username"
+        message <- obj .: "msg_id"
+        session <- obj .: "session"
+        return (messType, username, message, session)
+
+      -- Get metadata as a simple map.
+      Just metadataMap = decode $ Lazy.fromStrict metadata :: Maybe (Map ByteString ByteString)
+
+
+parseMessage :: [ByteString] -> ByteString -> ByteString -> ByteString -> ByteString -> Message
+parseMessage idents headerData parentHeader metadata content = 
+  let header = parseHeader idents headerData parentHeader metadata
+      body = parseMessageContent (msgType header) content in
+    Message header body
+
+parseMessageContent :: MessageType -> ByteString -> MessageBody
+parseMessageContent "kernel_info_request" _ = KernelInfoRequest
+parseMessageContent other _ = error $ "Unknown message type " ++ textToString (show other)
+

IHaskell/Types.hs

+{-# LANGUAGE TemplateHaskell #-}
+module IHaskell.Types (
+  Profile (..),
+  Message (..),
+  MessageHeader (..),
+  MessageBody (..),
+  MessageType,
+  Username,
+  Metadata,
+  Port,
+  messageTypeForBody
+  ) where
+
+import BasicPrelude
+
+import Data.Aeson
+import Data.Aeson.TH (deriveJSON)
+import Data.String.Utils (replace)
+import Data.UUID (UUID)
+import qualified Data.UUID as UUID (fromString, toString)
+
+instance FromJSON UUID where
+  parseJSON val@(String _) = do
+    str <- parseJSON val
+    case UUID.fromString str of
+      Nothing -> fail $ "Could not parse UUID from " ++ str
+      Just uuid -> return uuid
+  parseJSON _ = mzero
+
+instance ToJSON UUID where
+  toJSON = String . fromString . UUID.toString
+
+-- A kernel profile
+type Port = Int
+data Profile = Profile {
+  ip :: String,
+  transport :: String,
+  stdinPort :: Port,
+  controlPort :: Port,
+  hbPort :: Port,
+  shellPort :: Port,
+  iopubPort :: Port,
+  key :: ByteString
+  }
+$(deriveJSON (replace "Port" "_port") ''Profile)
+
+data MessageHeader = MessageHeader {
+  identifiers :: [ByteString],
+  parentHeader :: Maybe MessageHeader,
+  metadata :: Metadata,
+  messageId :: UUID,
+  sessionId :: UUID,
+  username :: Username,
+  msgType :: MessageType
+  } deriving Show
+
+instance ToJSON MessageHeader where
+  toJSON header = object [
+                    "msg_id"  .= messageId header,
+                    "session" .= sessionId header,
+                    "username" .= username header,
+                    "msg_type" .= msgType header
+                  ]
+
+data Message = Message {
+  header :: MessageHeader,
+  body :: MessageBody
+  } deriving Show
+data MessageBody = KernelInfoRequest |
+                   KernelInfoReply |
+                   ExecuteRequest {
+                     getCode :: ByteString,
+                     isSilent :: Bool,
+                     storeHistory :: Bool,
+                     allowStdin :: Bool,
+                     getUserVariables :: ByteString,
+                     getUserExpressions :: ByteString
+                   } deriving Show
+
+messageTypeForBody :: MessageBody -> ByteString
+messageTypeForBody KernelInfoRequest = "kernel_info_request"
+messageTypeForBody KernelInfoReply = "kernel_info_reply"
+messageTypeForBody _ = error "Unknown message type"
+
+instance ToJSON MessageBody where
+  toJSON KernelInfoReply = object [
+                             "protocol_version" .= [4, 0 :: Int], -- current protocol version, major and minor
+                             "language_version" .= [7, 6, 2 :: Int],
+                             "language" .= ("haskell" :: String)
+                           ]
+  toJSON body = error $ "Do not know how to convert to JSON for message" ++ textToString (show body)
+type MessageType = ByteString
+
+type Username = ByteString
+type Metadata = Map ByteString ByteString

Main.hs

+import BasicPrelude
+import Control.Concurrent
+import Control.Monad.Trans.State
+import Data.Aeson
+import Data.UUID.V4 (nextRandom)
+import System.Posix.Process
+import System.ZMQ
+import Text.Printf
+
+import qualified Data.ByteString.Lazy as ByteString
+import qualified Data.Map as Map
+
+import IHaskell.Types
+import IHaskell.MessageParser
+
+
+stdin :: Int
+stdin = 5678
+
+kernelProfile :: Profile
+kernelProfile = Profile {
+  ip = "127.0.0.1",
+  transport = "tcp",
+  stdinPort = stdin,
+  controlPort = stdin+1,
+  hbPort = stdin+2,
+  shellPort = stdin+3,
+  iopubPort = stdin+4,
+  key = ""
+  }
+
+main ::  IO ()
+main = do
+  putStrLn "Arguments:"
+  getArgs >>= print 
+
+  pid <- getProcessID
+  let fname = printf "profile-%s.json" $ textToString $ show pid
+  ByteString.writeFile fname $ encode kernelProfile
+  withContext 1 $ serveProfile kernelProfile
+
+serveProfile :: Profile -> Context -> IO ()
+serveProfile profile context = do
+ waitVar <- newEmptyMVar 
+ commandChannel <- newChan
+ replyChan <- newChan
+ putStrLn "Starting server..."
+ serveSocket context Router (stdinPort profile) stdinAction
+ serveSocket context Router (controlPort profile) $ shellAction commandChannel replyChan
+ serveSocket context Rep    (hbPort profile) hbAction
+ serveSocket context Router (shellPort profile) $ shellAction commandChannel replyChan
+ serveSocket context Pub    (iopubPort profile) iopubAction
+ putStrLn "Serving...."
+
+ let kernelState = 0
+     writeReply request reply = do
+       messageId <- nextRandom
+       let parent = header request
+           msgHead = MessageHeader {
+         identifiers = identifiers parent,
+         parentHeader = Just parent,
+         metadata = Map.fromList [],
+         messageId = messageId,
+         sessionId = sessionId parent,
+         username = username parent,
+         msgType = messageTypeForBody reply
+         }
+       let msg = Message {
+         header = msgHead,
+         body = reply
+         }
+       writeChan replyChan msg
+     evalMessage :: Int -> Message -> IO Int
+     evalMessage st message =
+       let (reply, newState) = runState (processMessage (body message)) st in do
+         putStr $ fromString $ printf "Responses: %d\n" newState 
+         writeReply message reply
+         return newState
+
+ messages <- getChanContents commandChannel
+ foldM_ evalMessage kernelState messages
+
+ takeMVar waitVar
+
+processMessage :: MessageBody -> State Int MessageBody
+processMessage KernelInfoRequest = do
+  modify (+1)
+  return KernelInfoReply
+processMessage msg = error $ "Unknown message type " ++ textToString (show msg) 
+  
+
+serveSocket :: SType a => Context -> a -> Port -> (Socket a -> IO b) -> IO ()
+serveSocket context socketType port action = void . forkIO $
+  withSocket context socketType $ \socket -> do
+    bind socket $ textToString $ "tcp://127.0.0.1:" ++ show port
+    forever $ action socket
+
+stdinAction :: Socket a -> IO ()
+stdinAction socket = do
+  request <- receive socket []
+  putStrLn "Received stdin."
+  print request
+
+hbAction :: Socket a -> IO ()
+hbAction socket = do
+  request <- receive socket []
+  putStrLn "Received heartbeat."
+  print request
+  -- echo back request
+  send socket request []
+
+shellAction :: Chan Message -> Chan Message -> Socket a -> IO ()
+shellAction channel readchan socket = do
+  msg <- ipythonReceive socket
+  writeChan channel msg
+  reply <- readChan readchan
+  ipythonSend socket reply
+
+iopubAction :: Socket a -> IO ()
+iopubAction socket = do
+  request <- receive socket []
+  putStrLn "Received iopub."
+  print request
+
+ipythonReceive :: Socket a -> IO Message
+ipythonReceive socket = do
+  let receiveNext = receive socket []
+      readMessage = do
+        line <- receiveNext
+        if line /= "<IDS|MSG>"
+        then do
+          next <- readMessage
+          return $ line : next
+        else return []
+  
+  idents <- readMessage
+  signature <- receive socket [] -- signature
+  headerData <- receive socket []
+  parentHeader <- receive socket []
+  metadata <- receive socket []
+  content <- receive socket []
+  putStrLn "ipython receive"
+  print idents
+  print signature
+  print headerData
+  print parentHeader
+  print metadata
+  print content
+  putStrLn "receive done"
+
+  return $ parseMessage idents headerData parentHeader metadata content
+
+ipythonSend :: Socket a -> Message -> IO ()
+ipythonSend socket message = 
+  let msgHead = header message
+      parentHeaderStr = case parentHeader msgHead of
+                          Nothing -> "{}"
+                          Just parentHead -> encode parentHead
+      idents = identifiers msgHead
+      metadata = "{}"
+      content = encode $ body message in do
+
+  let headerStr = ByteString.toStrict $ encode msgHead
+  putStrLn "ipython send"
+  forM_ idents $ \ident -> print ident
+  putStrLn "<IDS|MSG>"
+  putStrLn ""
+  print headerStr
+  print (ByteString.toStrict parentHeaderStr)
+  print metadata
+  print (ByteString.toStrict content)
+  putStrLn "send done"
+
+  forM_ idents $ \ident -> send socket ident [SndMore]
+  send socket "<IDS|MSG>" [SndMore]
+  send socket "" [SndMore]
+  send socket headerStr [SndMore]
+  send socket (ByteString.toStrict parentHeaderStr) [SndMore]
+  send socket metadata [SndMore]
+  send socket (ByteString.toStrict content) []
+

Setup.hs

+import Distribution.Simple
+
+main ::  IO ()
+main = defaultMain

interface.notes

+I have five sockets:
+    - Heartbeat
+    - Shell
+    - Control
+    - Publish
+    - Raw Input
+
+They are slightly documented here:
+  http://ipython.org/ipython-doc/stable/development/messaging.html#introduction
+
+
+Heartbeat:
+  Summary: Echo anything it gets sent.
+
+  IPython uses this to check that the kernel is still alive.
+
+Shell and Control:
+  Summary: Receive messages from frontend in specific format. Respond to messages.
+
+  This single ROUTER socket allows multiple incoming connections from frontends,
+  and this is the socket where requests for code execution, object information,
+  prompts, etc. are made to the kernel by any frontend. The communication on this
+  socket is a sequence of request/reply actions from each frontend and the
+  kernel.
+
+Publish (iopub):
+  Summary: Kernel sends side effects to front ends.
+
+  This socket is the ‘broadcast channel’ where the kernel publishes all side
+  effects (stdout, stderr, etc.) as well as the requests coming from any client
+  over the shell socket and its own requests on the stdin socket. There are a
+  number of actions in Python which generate side effects: print() writes to
+  sys.stdout, errors generate tracebacks, etc. Additionally, in a multi-client
+  scenario, we want all frontends to be able to know what each other has sent to
+  the kernel (this can be useful in collaborative scenarios, for example). This
+  socket allows both side effects and the information about communications taking
+  place with one client over the shell channel to be made available to all
+  clients in a uniform manner.
+
+Raw Input (stdin):
+  Summary: Allows kernel to request raw stdin from the frontent.
+
+  This ROUTER socket is connected to all frontends, and it allows the kernel to
+  request input from the active frontend when raw_input() is called. The frontend
+  that executed the code has a DEALER socket that acts as a ‘virtual keyboard’
+  for the kernel while this communication is happening (illustrated in the figure
+  by the black outline around the central keyboard). In practice, frontends may
+  display such kernel requests using a special input widget or otherwise
+  indicating that the user is to type input for the kernel instead of normal
+  commands in the frontend.