<commit_before>{-# LANGUAGE OverloadedStrings #-}
module Trombone.Dispatch.Pipeline where

import Control.Applicative                             ( (<$>) )
import Data.Conduit
import Data.Scientific
import Network.HTTP.Types                              
import Trombone.Dispatch.Core
import Trombone.Dispatch.Db                            ( escVal', dispatchDbAction )
import Trombone.Pipeline
import Trombone.RoutePattern

import qualified Data.Conduit.List                     as CL
import qualified Data.HashMap.Strict                   as HMS
import qualified Data.Text                             as Text
import qualified Data.Vector                           as Vect

dispatchPipeline :: Pipeline -> [(Text, EscapedText)] -> Value -> Dispatch RouteResponse
dispatchPipeline (Pipeline pcs conns _) ps obj = do
    mq <- initMq
    stabilize (Pipeline pcs conns mq) 
  where initMq = do
            Context{ dispatchRequest = r } <- ask
            -- Set up preliminary message queue from incoming connections
            return $ broadcast (outgoingConns In conns) $ foldr inject obj ps
        inject :: (Text, EscapedText) -> Value -> Value
        inject   (k, EscapedText v) (Object o) = Object $ HMS.insert k (String v) o
        inject p@(k, EscapedText v) (Array  a) = Array  $ Vect.map (inject p) a
        inject _                            x  = x

stabilize :: Pipeline -> Dispatch RouteResponse
stabilize sys@(Pipeline _ _ mq) = do
    Context pool _ _ _ _ loud <- ask
    when loud $ liftIO $ print mq
    s@(Pipeline _ _ mq') <- integrate sys 
    if mq == mq'  -- Has the message queue changed?
        then case processorMsgs Out mq of
               [] -> return $ errorResponse ErrorServerGeneric 
                                "Bad pipeline. (Null response)"
               msgs -> let o = combine msgs
                        in return $ RouteResponse []
                            (returnCode $ HMS.lookup "responseCode" o) 
                            (Object o)
        else stabilize s
  where returnCode (Just (Number n)) = 
            case floatingOrInteger n of
              Left  r -> 200
              Right i -> fromIntegral i
        returnCode _ = 200

combine :: [Message] -> Object
combine = foldr f HMS.empty
  where f (Message _ o1) o2 = 
            case HMS.toList o1 of
              [(key, val)] -> 
                  let ins a = HMS.insert key (Array $ Vect.cons val a) o2 in
                  case HMS.lookup key o2 of
                    Just (Array a) -> ins a 
                    Just v         -> ins (Vect.singleton v) 
                    Nothing -> o1 `HMS.union` o2
              _             -> o1 `HMS.union` o2
   
-- | Run a single integration step on a pipeline.
integrate :: Pipeline -> Dispatch Pipeline
integrate s@(Pipeline _ _ []) = return s
integrate (Pipeline pcs conns mq) = 
    liftM (Pipeline pcs conns . (++) (processorMsgs Out mq) 
                              . concat) (mapM f pcs)
  where f p = let pid = Id $ processorId p
                  msgs = processorMsgs pid mq in
                runProcessor p msgs $ outgoingConns pid conns

runProcessor :: Processor -> [Message] -> [Connection] -> Dispatch [Message]
runProcessor (Processor pid fields mtd uri exp) msgs conns = do
    let o = buildJsonRequest msgs
        v = expand exp o
    case (saturated fields v, fill uri o) of
        (True, Just x) -> do
            r <- lookupRoute mtd x
            case r of
                Nothing -> return []
                Just (Route _ _ (RouteSql    q), ps) -> do
                    RouteResponse _ _ r <- dispatchDbAction q ps v
                    return $ broadcast conns r
                Just (Route _ _ (RoutePipes  _), ps) -> return [] -- @todo
                Just (Route _ _ (RouteNodeJs _), ps) -> return [] -- @todo
                Just (Route _ _ (RouteStatic _), ps) -> return [] -- @todo
        _ -> return msgs

fill :: RoutePattern -> Object -> Maybe Text
fill (RoutePattern rp) o = Text.intercalate "/" <$> f rp []
  where f [] ps = Just ps
        f (Atom a:xs) ps = f xs (a:ps)
        f (Variable v:xs) ps = 
            case HMS.lookup v o of
                Nothing -> Nothing
                Just x  -> f xs $ escVal' x:ps
        
saturated :: [Text] -> Value -> Bool<commit_msg>fixed uri fill function in Dispatch/Pipeline.hs<commit_after>{-# LANGUAGE OverloadedStrings #-}
module Trombone.Dispatch.Pipeline where

import Control.Applicative                             ( (<$>) )
import Data.Conduit
import Data.Scientific
import Data.Text                                       ( isPrefixOf )
import Network.HTTP.Types                              
import Trombone.Dispatch.Core
import Trombone.Dispatch.Db                            ( dispatchDbAction )
import Trombone.Pipeline
import Trombone.RoutePattern

import qualified Data.Conduit.List                     as CL
import qualified Data.HashMap.Strict                   as HMS
import qualified Data.Text                             as Text
import qualified Data.Vector                           as Vect

dispatchPipeline :: Pipeline -> [(Text, EscapedText)] -> Value -> Dispatch RouteResponse
dispatchPipeline (Pipeline pcs conns _) ps obj = do
    mq <- initMq
    stabilize (Pipeline pcs conns mq) 
  where initMq = do
            Context{ dispatchRequest = r } <- ask
            -- Set up preliminary message queue from incoming connections
            return $ broadcast (outgoingConns In conns) $ foldr inject obj ps
        inject :: (Text, EscapedText) -> Value -> Value
        inject   (k, EscapedText v) (Object o) = Object $ HMS.insert k (String v) o
        inject p@(k, EscapedText v) (Array  a) = Array  $ Vect.map (inject p) a
        inject _                            x  = x

stabilize :: Pipeline -> Dispatch RouteResponse
stabilize sys@(Pipeline _ _ mq) = do
    Context pool _ _ _ _ loud <- ask
    when loud $ liftIO $ print mq
    s@(Pipeline _ _ mq') <- integrate sys 
    if mq == mq'  -- Has the message queue changed?
        then case processorMsgs Out mq of
               [] -> return $ errorResponse ErrorServerGeneric 
                                "Bad pipeline. (Null response)"
               msgs -> let o = combine msgs
                        in return $ RouteResponse []
                            (returnCode $ HMS.lookup "responseCode" o) 
                            (Object o)
        else stabilize s
  where returnCode (Just (Number n)) = 
            case floatingOrInteger n of
              Left  r -> 200
              Right i -> fromIntegral i
        returnCode _ = 200

combine :: [Message] -> Object
combine = foldr f HMS.empty
  where f (Message _ o1) o2 = 
            case HMS.toList o1 of
              [(key, val)] -> 
                  let ins a = HMS.insert key (Array $ Vect.cons val a) o2 in
                  case HMS.lookup key o2 of
                    Just (Array a) -> ins a 
                    Just v         -> ins (Vect.singleton v) 
                    Nothing -> o1 `HMS.union` o2
              _             -> o1 `HMS.union` o2
   
-- | Run a single integration step on a pipeline.
integrate :: Pipeline -> Dispatch Pipeline
integrate s@(Pipeline _ _ []) = return s
integrate (Pipeline pcs conns mq) = 
    liftM (Pipeline pcs conns . (++) (processorMsgs Out mq) 
                              . concat) (mapM f pcs)
  where f p = let pid = Id $ processorId p
                  msgs = processorMsgs pid mq in
                runProcessor p msgs $ outgoingConns pid conns

runProcessor :: Processor -> [Message] -> [Connection] -> Dispatch [Message]
runProcessor (Processor pid fields mtd uri exp) msgs conns = do
    Context _ _ _ _ _ loud <- ask
    let o = buildJsonRequest msgs
        v = expand exp o
    case (saturated fields v, fill uri o) of
        (True, Just x) -> do
            when loud $ liftIO $ putStrLn (show mtd ++ " " ++ Text.unpack x)
            r <- lookupRoute mtd x
            case r of
                Nothing -> return []
                Just (Route _ _ (RouteSql    q), ps) -> do
                    RouteResponse _ _ r <- dispatchDbAction q ps v
                    return $ broadcast conns r
                Just (Route _ _ (RoutePipes  _), ps) -> return [] -- @todo
                Just (Route _ _ (RouteNodeJs _), ps) -> return [] -- @todo
                Just (Route _ _ (RouteStatic _), ps) -> return [] -- @todo
        _ -> return msgs

fill :: RoutePattern -> Object -> Maybe Text
fill (RoutePattern rp) o = Text.intercalate "/" <$> f rp []
  where f [] ps = Just $ reverse ps
        f (Atom a:xs) ps = f xs (a:ps)
        f (Variable v:xs) ps = 
            case HMS.lookup (Text.cons ':' v) o of
                Just (String x)  -> f xs (strip x:ps) 
                _                -> Nothing

strip :: Text -> Text
strip = s . Text.reverse . s . Text.reverse 
  where s t |  "'" `isPrefixOf` t = Text.tail t
            | otherwise           = t

saturated :: [Text] -> Value -> Bool
{'fraction_non_alphanumeric': 0.08701949860724234, 'fraction_numerical': 0.003342618384401114, 'mean_word_length': 1.5536273115220485, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 74, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3075791', 'n_tokens_mistral': 2541, 'n_tokens_neox': 2425, 'n_words': 886}