miTask.icl

   1 module miTask
   2
   3 import StdDebug, StdMisc
   4
   5 from Text import class Text(concat,join,split), instance Text String
   6
   7 import iTasks
   8 import mTask
   9
  10 Start :: *World -> *World
  11 Start world = startEngine mTaskTask world
  12
  13 mTaskTask :: Task ()
  14 mTaskTask = withShared ([],False,[],False) (\ch->
  15                 syncNetworkChannel "localhost" 8123 "\n" id id ch ||-
  16                 updateSharedInformation "channels" [] ch @! ())
  17
  18 syncNetworkChannel :: String Int String (String -> m) (m -> String) (Shared ([m],Bool,[m],Bool)) -> Task () | iTask m
  19 syncNetworkChannel server port msgSeparator decodeFun encodeFun channel
  20         = tcpconnect server port channel {ConnectionHandlers|onConnect=onConnect,whileConnected=whileConnected,onDisconnect=onDisconnect} @! ()
  21         where
  22                 onConnect _ (received,receiveStopped,send,sendStopped)
  23                         = (Ok "",if (not (isEmpty send)) (Just (received,False,[],sendStopped)) Nothing, map encodeFun send,False)
  24                 whileConnected Nothing acc (received,receiveStopped,send,sendStopped)
  25                         = (Ok acc, Nothing, [], False)
  26                 whileConnected (Just newData) acc (received,receiveStopped,send,sendStopped)
  27                 # [acc:msgs] = reverse (split msgSeparator (concat [acc,newData]))
  28                 # write = if (not (isEmpty msgs && isEmpty send))
  29                         (Just (received ++ map decodeFun (reverse msgs),receiveStopped,[],sendStopped))
  30                         Nothing
  31                 = (Ok acc,write,map encodeFun send,False)
  32
  33                 onDisconnect l (received,receiveStopped,send,sendStopped)
  34                         = (Ok l,Just (received,True,send,sendStopped))
  35
  36 consumeNetworkStream :: ([m] -> Task ()) (Shared ([m],Bool,[m],Bool)) -> Task () | iTask m
  37 consumeNetworkStream processTask channel
  38         = ((watch channel >>* [OnValue (ifValue ifProcess process)]) <! id) @! ()
  39         where
  40                 ifProcess (received,receiveStopped,_,_)
  41                         = receiveStopped || (not (isEmpty received))
  42
  43                 process (received,receiveStopped,_,_)
  44                         =   upd empty channel
  45                         >>| if (isEmpty received) (return ()) (processTask received)
  46                         @!  receiveStopped
  47
  48                 empty :: ([m],Bool,[m],Bool) -> ([m],Bool,[m],Bool)
  49                 empty (_,rs,s,ss) = ([],rs,s,ss)