Internals.oz

   1 functor
   2 import
   3    Pickle(pack:Pack unpack:Unpack)
   4    Connection
   5    Open
   6    OS
   7    SerializationUtils(toAlpha:ToAlpha
   8                       completeTokens:CompleteTokens)
   9    DistFramework(session:Session)
  10 export
  11    InternalLayerPort
  12    InternalLayerSocket
  13 define
  14    fun{InternalLayerPort LM ?Uri}
  15       Tail
  16       ThisDest={Port.new {Cell.new $ Tail}}
  17       thread
  18          proc{Loop}
  19             Tail:=@Tail.2
  20             {Loop}
  21          end in
  22          {Loop}
  23       end
  24    in
  25       Uri='dist-base:internals'
  26       class from Session
  27          meth init()
  28             skip
  29          end
  30          meth thisDest($)
  31             ThisDest
  32          end
  33          meth serializeDest(D $)
  34             {Connection.offerMany D}
  35          end
  36          meth deserializeDest(D $)
  37             {Connection.take D}
  38          end
  39          meth send(D M)
  40             {Port.send D iMsg(ThisDest M)}
  41          end
  42          meth receive($)
  43             @Tail
  44          end
  45       end
  46    end
  47    fun{InternalLayerSocket LM ?Uri}
  48       fun{BetterIP IP1 IP2}
  49          for
  50             P in ["127." "169.254." "192.168."
  51                   "172.16." "172.17." "172.18." "172.19."
  52                   "172.20." "172.21." "172.22." "172.23."
  53                   "172.24." "172.25." "172.26." "172.27."
  54                   "172.28." "172.29." "172.30." "172.31."
  55                   "10."]
  56             return:R
  57          do
  58             if {List.isPrefix P IP1} then {R false} end
  59             if {List.isPrefix P IP2} then {R true} end
  60          end
  61       end
  62       fun{GetHost}
  63          IPs={Append
  64               {OS.getHostByName "localhost"}.addrList
  65               {OS.getHostByName {OS.uName}.nodename}.addrList
  66              } in
  67          {List.sort IPs BetterIP}.1
  68       end
  69       Tail
  70       RPort={Port.new {Cell.new $ Tail}}
  71       thread
  72          proc{Loop}
  73             Tail:=@Tail.2
  74             {Loop}
  75          end in
  76          {Loop}
  77       end
  78       ServSocket={New Open.socket init()}
  79       TCPPort={ServSocket bind(port:$)}
  80       ThisDest=site(id:{NewName} host:{GetHost} port:TCPPort)
  81       {ServSocket listen()}
  82       thread
  83          proc{Loop}
  84             Bytes Blobs Msgs
  85             ThatSocket={ServSocket
  86                         accept(accepted:$
  87                                acceptClass:Open.socket)}
  88          in
  89             thread
  90                %%{Browse bytes(Bytes)}
  91                {ThatSocket read(list:Bytes size:all)}
  92             end
  93             thread
  94                %%{Browse blobs(Blobs)}
  95                Blobs={CompleteTokens Bytes [&,]}
  96             end
  97             thread
  98                %%{Browse msgs(Msgs)}
  99                Msgs={Map Blobs Unpack}
 100             end
 101             thread
 102                Src=Msgs.1 in
 103                for M in Msgs.2 do
 104                   {Port.send RPort iMsg(Src M)}
 105                end
 106             end
 107             {Loop}
 108          end
 109       in
 110          {Loop}
 111       end
 112       Connections={Dictionary.new}
 113    in
 114       Uri='dist-base:internals'
 115       class from Session
 116          meth init()
 117             skip
 118          end
 119          meth thisDest($)
 120             ThisDest
 121          end
 122          meth serializeDest(D $)
 123             {Pickle.pack D}
 124          end
 125          meth deserializeDest(D $)
 126             {Pickle.unpack D}
 127          end
 128          meth send(D M)
 129             P in
 130             case {Dictionary.condExchange
 131                   Connections D.id unit $ P}
 132             of unit then
 133                ThatSocket={New Open.socket client(host:D.host
 134                                                   port:D.port)}
 135             in
 136                {ThatSocket
 137                 write(vs:{ToAlpha {Pack ThisDest} ","})}
 138                P={Port.new
 139                   thread
 140                      for M in $ do
 141                         %%{Browse msg(M)}
 142                         {ThatSocket
 143                          write(vs:{ToAlpha {Pack M} ","})}
 144                      end
 145                   end}
 146             [] O then
 147                P=O
 148             end
 149             {Port.send P M}
 150          end
 151          meth receive($)
 152             @Tail
 153          end
 154       end
 155    end
 156 end