csharp/dotnet/pxprpc/ServerContext.cs

   1 using System;
   2 using System.Collections.Generic;
   3 using System.IO;
   4 using System.Text;
   5 using System.Threading;
   6
   7 namespace pxprpc
   8 {
   9     public class ServerContext
  10     {
  11         public Stream stream;
  12         public static int DefaultRefSlotsCap = 256;
  13         public PxpObject[] refSlots = new PxpObject[256];
  14         public Dictionary<String, Object> funcMap = new Dictionary<String, Object>();
  15         protected BuiltInFuncList builtIn;
  16         public void init(Stream stream)
  17         {
  18             this.stream = stream;
  19             builtIn = new BuiltInFuncList();
  20             funcMap["builtin"] = builtIn;
  21         }
  22         public bool running = true;
  23
  24         protected void putRefSlots(int addr, PxpObject r)
  25         {
  26             if (refSlots[addr] != null)
  27                 refSlots[addr].release();
  28             if (r != null)
  29                 r.addRef();
  30             refSlots[addr] = r;
  31         }
  32         private Mutex priv__writeLock = new Mutex();
  33         public Mutex writeLock()
  34         {
  35             return priv__writeLock;
  36         }
  37         public void push(PxpRequest r)
  38         {
  39             putRefSlots(r.destAddr, new PxpObject(r.parameter));
  40             writeLock().WaitOne();
  41             this.stream.Write(r.session);
  42             writeLock().ReleaseMutex();
  43             this.stream.Flush();
  44         }
  45         public void pull(PxpRequest r)
  46         {
  47             Object o = null;
  48             if (refSlots[r.srcAddr] != null)
  49             {
  50                 o = refSlots[r.srcAddr].get();
  51             }
  52             writeLock().WaitOne();
  53             this.stream.Write(r.session);
  54             if (o is byte[])
  55             {
  56                 byte[] b = (byte[])o;
  57                 writeInt32(b.Length);
  58                 stream.Write(b);
  59             }
  60             else if (o is String)
  61             {
  62                 byte[] b = Encoding.UTF8.GetBytes((String)o);
  63                 writeInt32(b.Length);
  64                 stream.Write(b);
  65             }
  66             else
  67             {
  68                 writeInt32(-1);
  69             }
  70             writeLock().ReleaseMutex();
  71             stream.Flush();
  72         }
  73         public void assign(PxpRequest r)
  74         {
  75             putRefSlots(r.destAddr, this.refSlots[r.srcAddr]);
  76             writeLock().WaitOne();
  77             this.stream.Write(r.session);
  78             writeLock().ReleaseMutex();
  79             this.stream.Flush();
  80         }
  81         public void unlink(PxpRequest r)
  82         {
  83             putRefSlots(r.destAddr, null);
  84             writeLock().WaitOne();
  85             this.stream.Write(r.session);
  86             writeLock().ReleaseMutex();
  87             this.stream.Flush();
  88         }
  89         public void call(PxpRequest r)
  90         {
  91             r.pending = true;
  92             r.callable.call(r, (Object result) =>
  93          {
  94
  95              r.result = result;
  96              putRefSlots(r.destAddr, new PxpObject(result));
  97              r.pending = false;
  98              writeLock().WaitOne();
  99              stream.Write(r.session);
 100              r.callable.writeResult(r);
 101              writeLock().ReleaseMutex();
 102              stream.Flush();
 103
 104          });
 105         }
 106
 107
 108         public void getInfo(PxpRequest r)
 109         {
 110             writeLock().WaitOne();
 111             this.stream.Write(r.session);
 112             byte[]
 113             b = Encoding.UTF8.GetBytes(
 114             "server name:pxprpc for c#\n" +
 115             "version:1.0\n" +
 116             "reference slots capacity:" + this.refSlots.Length + "\n"
 117             );
 118             writeInt32(b.Length);
 119             stream.Write(b);
 120             writeLock().ReleaseMutex();
 121             stream.Flush();
 122         }
 123
 124         public void serve()
 125         {
 126             this.running = true;
 127             while (running)
 128             {
 129                 PxpRequest r = new PxpRequest();
 130                 r.context = this;
 131                 byte[] session = new byte[4];
 132                 Utils.readf(stream, session);
 133                 r.session = session;
 134                 r.opcode = session[0];
 135                 switch (r.opcode)
 136                 {
 137                     case 1:
 138                         r.destAddr = readInt32();
 139                         int len = readInt32();
 140                         byte[] buf = new byte[len];
 141                         Utils.readf(stream, buf);
 142                         r.parameter = buf;
 143                         push(r);
 144                         break;
 145                     case 2:
 146                         r.srcAddr = readInt32();
 147                         pull(r);
 148                         break;
 149                     case 3:
 150                         r.destAddr = readInt32();
 151                         r.srcAddr = readInt32();
 152                         assign(r);
 153                         break;
 154                     case 4:
 155                         r.destAddr = readInt32();
 156                         unlink(r);
 157                         break;
 158                     case 5:
 159                         r.destAddr = readInt32();
 160                         r.srcAddr = readInt32();
 161                         r.callable = (PxpCallable)refSlots[r.srcAddr].get();
 162                         r.callable.readParameter(r);
 163                         call(r);
 164                         break;
 165                     case 6:
 166                         r.destAddr = readInt32();
 167                         r.srcAddr = readInt32();
 168                         getFunc(r);
 169                         break;
 170                     case 7:
 171                         Dispose();
 172                         running = false;
 173                         break;
 174                     case 8:
 175                         getInfo(r);
 176                         break;
 177                 }
 178             }
 179             running = false;
 180         }
 181         public int readInt32()
 182         {
 183             return Utils.readInt32(stream);
 184         }
 185
 186         public long readInt64()
 187         {
 188             return Utils.readInt64(stream);
 189         }
 190         public float readFloat32()
 191         {
 192             return Utils.readFloat32(stream);
 193         }
 194         public double readFloat64()
 195         {
 196             return Utils.readFloat64(stream);
 197         }
 198         public byte[] readNextRaw()
 199         {
 200             int addr = readInt32();
 201             return (byte[])refSlots[addr].get();
 202         }
 203         public void getFunc(PxpRequest r)
 204         {
 205             String name = getStringAt(r.srcAddr);
 206             int namespaceDelim = name.LastIndexOf(".");
 207             String ns = name.Substring(0, namespaceDelim);
 208             String func = name.Substring(namespaceDelim + 1);
 209             Object obj = funcMap[ns];
 210             PxpCallable found = null;
 211             if (obj != null)
 212             {
 213                 found = builtIn.getBoundMethod(obj, func);
 214             }
 215             writeLock().WaitOne();
 216             if (found == null)
 217             {
 218
 219                 this.stream.Write(r.session);
 220
 221                 writeInt32(0);
 222             }
 223             else
 224             {
 225                 putRefSlots(r.destAddr, new PxpObject(found));
 226
 227                 this.stream.Write(r.session);
 228
 229                 writeInt32(r.destAddr);
 230
 231             }
 232             writeLock().ReleaseMutex();
 233             this.stream.Flush();
 234         }
 235
 236         public String readNextString()
 237         {
 238             int addr = readInt32();
 239             return getStringAt(addr);
 240         }
 241
 242         public String getStringAt(int addr)
 243         {
 244             if (refSlots[addr] == null)
 245             {
 246                 return "";
 247             }
 248             Object o = refSlots[addr].get();
 249             if (o is byte[])
 250             {
 251                 return Encoding.UTF8.GetString((byte[])o);
 252             }
 253             else
 254             {
 255                 return o.ToString();
 256             }
 257         }
 258
 259         public void writeInt32(int d)
 260         {
 261             Utils.writeInt32(stream, d);
 262         }
 263         public void writeInt64(long d)
 264         {
 265             Utils.writeInt64(stream, d);
 266         }
 267         public void writeFloat32(float d)
 268         {
 269             Utils.writeFloat32(stream, d);
 270         }
 271         public void writeFloat64(double d)
 272         {
 273             Utils.writeFloat64(stream, d);
 274         }
 275
 276
 277         public void Dispose()
 278         {
 279             running = false;
 280             stream.Close();
 281             stream.Dispose();
 282             for (int i1 = 0; i1 < refSlots.Length; i1++)
 283             {
 284                 if (refSlots[i1] != null)
 285                 {
 286                     try
 287                     {
 288                         refSlots[i1].release();
 289                     }
 290                     catch (Exception e) { }
 291                 }
 292             }
 293         }
 294     }
 295 }