DistFramework.oz

   1 functor
   2 import
   3    LKDictionary(new:NewLKDic)
   4    Property
   5    Comm
   6    Dist
   7 export
   8    Session
   9    LayerManager
  10    Version
  11 define
  12    Version=4
  13    local
  14       SessionInit={NewName}
  15       PreInit={NewName}
  16    in
  17       class LayerManagerClass
  18          attr
  19             layers
  20          meth !PreInit()
  21             layers:=nonInitialized
  22          end
  23          meth init(Kind<=port)
  24             layers:={Dictionary.new}
  25             {self introduceLayer(AlarmLayer)}
  26             for F in [remoteProcess router localProcess] do
  27                {self introduceLayer(Comm.F)}
  28             end
  29             for F in [comm process bep2p pp2p ifd pfd] do
  30                {self introduceLayer(Dist.F)}
  31             end
  32          end
  33          meth introduceLayer(LayerFunc)
  34             LayerUri
  35             LayerClass={LayerFunc self LayerUri}
  36          in
  37             fun{{self getLayer(LayerUri $)} UserInit}
  38                {New LayerClass SessionInit(UserInit $) _}
  39             end
  40          end
  41          meth getLayer(Uri ?N)
  42             {Dictionary.condExchange @layers Uri _ N N}
  43          end
  44       end
  45       LayerManager={New LayerManagerClass PreInit()}
  46       class Session
  47          attr this
  48          meth !SessionInit(UserInit ?This)
  49             P in
  50             proc{This M} {@this M} end
  51             this:=proc{$ M} {Send P M} end
  52             P={NewPort thread
  53                           {self UserInit}
  54                           for M in $ do
  55                              {self M}
  56                           end
  57                        end}
  58          end
  59          meth facet($ ...)=S
  60             proc{$ M}
  61                L={Label M} in
  62                if {HasFeature S L} then
  63                   {@this {Adjoin M S.L}}
  64                end
  65             end
  66          end
  67       end
  68    end
  69    fun{AlarmLayer LM ?Uri}
  70       Alarms={NewCell nil}
  71       Sessions={NewLKDic}
  72       proc{Tick}
  73          ToDo Now OAl NAl
  74       in
  75          OAl=Alarms:=NAl
  76          {Wait OAl}
  77          Now={Property.get 'time.total'}
  78          NAl={List.takeDropWhile OAl fun{$ T#_#_}T<Now end ToDo $}
  79          for _#S#P in ToDo do
  80             {{Sessions get(S $)} alarm(P)}
  81          end
  82          if NAl\=nil then
  83             {Delay {Max 0 Now-NAl.1.1}}
  84             {Tick}
  85          end
  86       end
  87    in
  88       Uri='dist-layer:alarm'
  89       class from Session
  90          attr sid
  91          meth init(SId Handler)
  92             sid:=SId
  93             {Sessions put(@sid Handler)}
  94          end
  95          meth getTime($)
  96             {Property.get 'time.total'}
  97          end
  98          meth setAlarmAt(Time Payload)
  99             OAl NAl in
 100             OAl=Alarms:=NAl
 101             NAl={List.merge OAl [Time#@sid#Payload] fun{$ X#_#_ Y#_#_}X<Y end}
 102             if OAl==nil orelse OAl.1.1 > Time then
 103                thread
 104                   {Delay {Max 0 Time-{self getTime($)}}}
 105                   {Tick}
 106                end
 107             end
 108          end
 109          meth setAlarmIn(Interval Payload)
 110             {self setAlarmAt({self getTime($)}+Interval Payload)}
 111          end
 112       end
 113    end
 114 end