此处案例将ICE接口当做单servant使用(ICE自带端口复用的多servant,过于复杂,此处不讨论)
使用ICE较为方便的地方时 可以编写 ice中间代码,然后由官方工具转换为目标平台代码(通过语句自动生成),生产的代码结构比较复杂,不赘述,此处需要注意一点,自动生成的文件夹路径不要包含特殊字符。
ICE的使用概念与传统的RPC框架没有差异。过程如下:
1、由某一方建立监听,作为通讯宿主(习惯称之为服务器)
2、由另外一方建立链接到服务器(习惯称之为客户端)
3、客户端向服务器注册自己的信息(可能会有多个客户端,意味着可以注册多个客户,服务器需要存储)
4、双向自由互调。
其中双向互调过程中,主动发起调用方名义上都叫做 client ,提供服务方 都叫做 servant。
双方需要有一个约定,即上面提到的自动生成的 <interface>Disp_ 抽象类,作为双方统一接口。双方均需要定义自己的服务去实现该抽象类(client/servant 作用不同,实现的方式自然不同)。
为了满足多个不同端口不通服务使用,笔者对客户端和服务器进行了简单封装,如下
需要改进重连机制
public class ICEClient<T,TPrx> { public TPrx Server { get; private set; } public bool Online { get; private set; } ILog _Logger = Spv2LoggerFactory.GetLoggerByName("RPC"); string _IPAddress; int _Port; int connectTimeout = 60000; int idleTimeout = -1; //no limit public ICEClient(string ip, int port) { _IPAddress = ip; _Port = port; } /// <summary> /// /// </summary> /// <param name="Servant"> 作为服务端接收指令响应</param> public void Connect(T Servant=default(T)) { if (Online) { _Logger.InfoFormat("Initializing, Online!"); return; } try { _Logger.InfoFormat("Initializing ICE Comm using Host: {0}, Port: {1}, connectTimeout: {2}, idleTimeout: {3}", _IPAddress, _Port, connectTimeout, idleTimeout); // TODO figure out how to use properties. Ice.InitializationData id = new Ice.InitializationData(); id.properties = Ice.Util.createProperties(); id.properties.setProperty("Ice.Default.ConnectTimeout", connectTimeout.ToString()); id.properties.setProperty("Ice.Default.Timeout", idleTimeout.ToString()); // 5 min? id.properties.setProperty("Ice.Warn.UnusedProperties", "1"); id.properties.setProperty("Ice.Warn.Host", _IPAddress); var _ICEComm = Ice.Util.initialize(id); string connectString = String.Format("{0}:tcp -p {1} -h {2}", typeof(T).Name, _Port, _IPAddress); ObjectPrx iceProxy = _ICEComm.stringToProxy(connectString); Server = CreateServerProxy(iceProxy); _Logger.InfoFormat("ICE check proxy cast finished. {0}", _IPAddress); if (Server == null) { _ICEComm.destroy(); _ICEComm = null; id = null; GC.Collect(); throw new System.Exception("Invalid proxy"); } if(Servant!=null) { Ice.ObjectAdapter iceAdapter = _ICEComm.createObjectAdapter(""); iceAdapter.add((Ice.Object)Servant, iceProxy.ice_getIdentity()); iceProxy.ice_getConnection().setAdapter(iceAdapter); iceAdapter.activate(); } } catch (System.Exception ex) { Online = false; GC.Collect(); string errorMsg = String.Format("System Exception {0} caught.", ex.Message); _Logger.Error(errorMsg); throw; } Online = true; } private TPrx CreateServerProxy(ObjectPrx iceProxy) { //TestICEPrxHelper var str = typeof(TPrx).FullName + "Helper"; var type = Type.GetType(str); var serverproxy = Activator.CreateInstance(type); //static method var method = type.GetRuntimeMethod("uncheckedCast", new Type[] { typeof(ObjectPrx) }); return (TPrx)method.Invoke(type, new[] { iceProxy }); } }
public class ICEServer<T,TPrx> { ILog _Logger = Spv2LoggerFactory.GetLoggerByName("RPC"); /// <summary> /// 客户端的集合 /// </summary> public Dictionary<string,TPrx> Clients { get; set; } = new Dictionary<string, TPrx>(); string _IPAddress; int _Port; public ICEServer(string ip, int port) { _IPAddress = ip; _Port = port; } /// <summary> /// /// </summary> /// <param name="Servant"> 作为服务端接收指令响应</param> public void StartUp(T Servant) { try { _Logger.InfoFormat("Initializing ICE Comm using Host: {0}, Port: {1}",_IPAddress, _Port); var _ICEComm = Ice.Util.initialize(); Ice.Communicator iceComm = Ice.Util.initialize(); // Create adaptor for commands and requests from client Ice.ObjectAdapter iceAdapter = iceComm.createObjectAdapterWithEndpoints(typeof(T).Name, $"tcp -p {_Port} -h {_IPAddress}"); iceAdapter.add((Ice.Object)Servant, iceComm.stringToIdentity(typeof(T).Name)); iceAdapter.activate(); } catch (System.Exception ex) { GC.Collect(); string errorMsg = String.Format("System Exception {0} caught.", ex.Message); _Logger.Error(errorMsg); throw; } } }
以下为测试代码
// Copyright 2016 Complete Genomics, Inc. All Rights Reserved. // Confidential and proprietary works of Complete Genomics, Inc. module My { module RPC { interface TestICE { void AddClient(string name); string ClientCallServer(int i); void ServerCallClient(int i); }; }; };
internal class TestICEServer : TestICEDisp_ { ICEServer<TestICE, TestICEPrx> server; internal TestICEServer() { server = new ICEServer<TestICE, TestICEPrx>("172.16.35.66", 11106); server.StartUp(this); } public override void AddClient(string name, Current current__) { Ice.ObjectPrx @base = current__.con.createProxy(current__.id); var client = TestICEPrxHelper.checkedCast(@base); server.Clients.Add(name, client); } public override string ClientCallServer(int i, Current current__) { return (++i).ToString(); } public override void ServerCallClient(int i, Current current__) { foreach (var client in server.Clients.Values) { client.ServerCallClient(i); } } } public class TestServerService : TestICEOperationsNC_ { TestICEServer server; public TestServerService() { server = new TestICEServer(); } public void AddClient(string name) { server.AddClient(name); } public string ClientCallServer(int i) { return server.ClientCallServer(i); } public void ServerCallClient(int i) { server.ServerCallClient(i); } }
internal class TestICEClient : TestICEDisp_ { ICEClient<TestICE, TestICEPrx> client; internal TestICEClient() { client = new ICEClient<TestICE, TestICEPrx>("172.16.35.66", 11106); client.Connect(this); } public override void AddClient(string name, Current current__) { client.Server.AddClient(name); } public override string ClientCallServer(int i, Current current__) { var r= client.Server.ClientCallServer(i); return r; } public override void ServerCallClient(int i, Current current__) { var r = i; } } public class TestICEClientService : TestICEOperationsNC_ { TestICEClient server; public TestICEClientService() { server = new TestICEClient(); } public void AddClient(string name) { server.AddClient(name); } public string ClientCallServer(int i) { return server.ClientCallServer(i); } public void ServerCallClient(int i) { server.ServerCallClient(i); } }