| /* |
| * |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| * |
| */ |
| |
| using System; |
| using System.Collections.Generic; |
| using System.IO ; |
| using System.Reflection ; |
| using System.Threading ; |
| using log4net ; |
| using org.apache.qpid.client ; |
| using org.apache.qpid.transport.util; |
| using org.apache.qpid.transport.codec ; |
| |
| namespace org.apache.qpid.console |
| { |
| |
| /** |
| * Main Interaction point for interaction with the bus. Can be used to locate objects |
| * on the bus, and invoke messages on them. |
| */ |
| public class Session |
| { |
| public static ILog log = LogManager.GetLogger(typeof(Session)) ; |
| |
| public static int CONTEXT_SYNC = 1 ; |
| public static int CONTEXT_STARTUP = 2 ; |
| public static int CONTEXT_MULTIGET = 3 ; |
| public static int DEFAULT_GET_WAIT_TIME = 60000 ; |
| |
| public bool RecieveObjects = true ; |
| public bool RecieveEvents = true ; |
| public bool RecieveHeartbeat = true ; |
| public bool UserBindings = false ; |
| public Console Console ; |
| |
| protected Dictionary<string, Dictionary<string, SchemaClass>> Packages = new Dictionary<string, Dictionary<string, SchemaClass>>() ; |
| protected List<Broker> Brokers = new List<Broker>() ; |
| protected SequenceManager SequenceManager = new SequenceManager() ; |
| protected object LockObject = new Object(); |
| protected List<long> SyncSequenceList = new List<long>() ; |
| protected List<QMFObject> GetResult ; |
| protected Object SyncResult ; |
| |
| public Session() |
| { |
| } |
| |
| public Session(Console console) { |
| Console = console ; |
| } |
| |
| public void AddBroker(string url) { |
| BrokerURL brokerurl = GetBrokerURL(url) ; |
| Broker broker = new Broker(this, brokerurl) ; |
| Brokers.Add(broker) ; |
| |
| Dictionary<string, object> args = new Dictionary<string, object>() ; |
| args.Add("_class", "agent") ; |
| args.Add("_broker", broker) ; |
| this.GetObjects(args) ; |
| } |
| |
| public void RemoveBroker(Broker broker) { |
| if (Brokers.Contains(broker)) { |
| Brokers.Remove(broker) ; |
| } |
| |
| broker.Shutdown() ; |
| } |
| |
| public void Close() { |
| foreach (Broker broker in Brokers.ToArray()) { |
| this.RemoveBroker(broker) ; |
| } |
| } |
| |
| protected BrokerURL GetBrokerURL(string url) { |
| return new BrokerURL(url) ; |
| } |
| |
| |
| public List<QMFObject> GetObjects(Dictionary<string, object> args) { |
| List<Broker> brokerList = null ; |
| List<Agent> agentList = new List<Agent>() ; |
| |
| if (args.ContainsKey("_broker")) { |
| brokerList = new List<Broker>() ; |
| brokerList.Add((Broker)args["_broker"]) ; |
| } else { |
| brokerList = this.Brokers ; |
| } |
| |
| foreach (Broker broker in brokerList) { |
| broker.WaitForStable() ; |
| } |
| |
| if (args.ContainsKey("_agent")) { |
| Agent agent = (Agent)args["_agent"] ; |
| if (brokerList.Contains(agent.Broker)) { |
| agentList.Add(agent) ; |
| } else { |
| throw new Exception("Agent is not managed by this console or the supplied broker") ; |
| } |
| } else { |
| if (args.ContainsKey("_objectId")) { |
| ObjectID oid = (ObjectID) args["_objectId"] ; |
| foreach (Broker broker in Brokers) { |
| foreach (Agent agent in broker.Agents.Values) { |
| if ((agent.AgentBank == oid.AgentBank()) && (agent.BrokerBank == oid.BrokerBank())) { |
| agentList.Add(agent) ; |
| } |
| } |
| } |
| } |
| else { |
| foreach (Broker broker in brokerList) { |
| foreach (Agent agent in broker.Agents.Values) { |
| if (agent.Broker.IsConnected()) { |
| agentList.Add(agent) ; |
| } |
| } |
| } |
| } |
| } |
| |
| GetResult = new List<QMFObject>() ; |
| |
| if (agentList.Count > 0) { |
| //FIXME Add a bunch of other suff too |
| foreach (Agent agent in agentList) { |
| Dictionary<string, object> getParameters = new Dictionary<string, object>() ; |
| Broker broker = agent.Broker ; |
| long seq = -1 ; |
| lock(LockObject) { |
| seq = SequenceManager.Reserve(Session.CONTEXT_MULTIGET) ; |
| SyncSequenceList.Add(seq) ; |
| } |
| object packageName = null ; |
| object className = null ; |
| object key = null ; |
| object sClass = null ; |
| object oid = null ; |
| object hash = null ; |
| |
| args.TryGetValue("_schema", out sClass) ; |
| args.TryGetValue("_key", out key) ; |
| args.TryGetValue("_class", out className) ; |
| args.TryGetValue("_package", out packageName) ; |
| args.TryGetValue("_objectID", out oid) ; |
| args.TryGetValue("_hash", out hash) ; |
| |
| if ((className == null) && (oid == null) && (oid == null)) { |
| throw new Exception("No class supplied, use '_schema', '_key', '_class', or '_objectId' argument") ; |
| } |
| |
| if (oid != null) { |
| getParameters.Add("_objectID", oid) ; |
| } |
| else { |
| if (sClass != null) { |
| key = key ?? ((SchemaClass)sClass).Key ; |
| } |
| if (key != null) { |
| ClassKey cKey = (ClassKey)key ; |
| className = className ?? cKey.ClassName ; |
| packageName = packageName ?? cKey.PackageName ; |
| hash = hash ?? cKey.Hash ; |
| } |
| |
| if (packageName != null) { |
| getParameters.Add("_package", packageName) ; |
| } |
| if (className != null) { |
| getParameters.Add("_class", className) ; |
| } |
| if (hash != null) { |
| getParameters.Add("_hash", hash) ; |
| } |
| foreach (KeyValuePair<string, object> pair in args) { |
| if (!pair.Key.StartsWith("_")) { |
| getParameters.Add(pair.Key, pair.Value) ; |
| } |
| } |
| } |
| |
| IEncoder enc = broker.CreateEncoder('G', seq) ; |
| enc.WriteMap(getParameters) ; |
| string routingKey = String.Format("agent.{0}.{1}", agent.BrokerBank, agent.AgentBank) ; |
| Message msg = broker.CreateMessage(enc, routingKey) ; |
| log.Debug("Get Object Keys: ") ; |
| foreach (string pKey in getParameters.Keys) { |
| log.Debug(String.Format("\tKey: '{0}' Value: '{1}'", pKey, getParameters[pKey])) ; |
| } |
| broker.Send(msg) ; |
| } |
| |
| int waittime = DEFAULT_GET_WAIT_TIME ; |
| bool timeout = false ; |
| if (args.ContainsKey("_timeout")) { |
| waittime = (int) args["_timeout"] ; |
| } |
| DateTime start = DateTime.Now ; |
| lock (LockObject) { |
| // FIXME ERROR |
| while (SyncSequenceList.Count > 0){ |
| Monitor.Wait(LockObject,waittime) ; |
| TimeSpan duration = DateTime.Now - start; |
| if (duration.TotalMilliseconds > waittime) { |
| foreach (long pendingSeq in SyncSequenceList) { |
| SequenceManager.Release(pendingSeq) ; |
| } |
| SyncSequenceList.Clear() ; |
| timeout = true ; |
| } |
| } |
| } |
| |
| //FIXME Add the error logic |
| if ((GetResult.Count == 0) && timeout) { |
| throw new Exception ("Get Request timed out") ; |
| } |
| } |
| return GetResult ; |
| } |
| |
| public List<string> GetPackages() { |
| this.WaitForStable() ; |
| List<string> returnValue = new List<string>() ; |
| foreach (String name in Packages.Keys) { |
| returnValue.Add(name) ; |
| } |
| |
| return returnValue ; |
| } |
| |
| public List<ClassKey> GetClasses(string packageName) { |
| List<ClassKey> returnValue = new List<ClassKey>() ; |
| this.WaitForStable() ; |
| |
| if (Packages.ContainsKey(packageName)) { |
| foreach (SchemaClass sClass in Packages[packageName].Values) { |
| returnValue.Add(sClass.Key) ; |
| } |
| } |
| |
| return returnValue ; |
| } |
| |
| public SchemaClass GetSchema(ClassKey key) { |
| return GetSchema(key, true) ; |
| } |
| |
| protected SchemaClass GetSchema(ClassKey key, bool waitForStable) { |
| if (waitForStable) { |
| this.WaitForStable() ; |
| } |
| |
| SchemaClass returnValue = null ; |
| |
| try { |
| returnValue = Packages[key.PackageName][key.GetKeyString()] ; |
| } |
| catch (KeyNotFoundException) { |
| // eat it |
| } |
| |
| return returnValue ; |
| } |
| |
| |
| protected void WaitForStable() { |
| foreach (Broker broker in Brokers) { |
| broker.WaitForStable() ; |
| } |
| } |
| |
| |
| public Broker GetBroker(long BrokerBank) { |
| Broker returnValue = null ; |
| |
| foreach (Broker broker in Brokers) { |
| if (broker.BrokerBank() == BrokerBank) { |
| returnValue = broker ; |
| break ; |
| } |
| } |
| |
| return returnValue ; |
| } |
| |
| public MethodResult InvokeMethod(QMFObject obj, string name, List<object> args, bool synchronous, int timeToLive) { |
| Broker aBroker = this.GetBroker(obj.BrokerBank()) ; |
| |
| long seq = this.SendMethodRequest(obj, aBroker, name, args, synchronous, timeToLive) ; |
| if (seq != 0) { |
| if (!synchronous) { |
| return null ; |
| } |
| try { |
| aBroker.WaitForSync(timeToLive) ; |
| } catch (Exception e) { |
| SequenceManager.Release(seq) ; |
| throw e ; |
| } |
| |
| // FIXME missing error logic in the broker |
| return (MethodResult) SyncResult ; |
| } |
| |
| return null ; |
| } |
| |
| protected long SendMethodRequest(QMFObject obj, Broker aBroker, string name, List<object> args, bool synchronous, int timeToLive) { |
| |
| SchemaMethod method = obj.Schema.GetMethod(name) ; |
| if (args == null) { |
| args = new List<object>() ; |
| } |
| |
| long seq = 0 ; |
| if (method != null) { |
| KeyValuePair<SchemaMethod, bool> pair = new KeyValuePair<SchemaMethod, bool>(method, synchronous) ; |
| seq = SequenceManager.Reserve(pair) ; |
| IEncoder enc = aBroker.CreateEncoder('M', seq) ; |
| obj.ObjectID.encode(enc) ; |
| obj.Schema.Key.encode(enc) ; |
| enc.WriteStr8(name) ; |
| |
| if (args.Count < method.InputArgCount) { |
| throw new Exception(String.Format("Incorrect number of arguments: expected {0}, got{1}", method.InputArgCount, args.Count)) ; |
| } |
| |
| int argIndex = 0 ; |
| foreach (SchemaArgument arg in method.Arguments) { |
| if (arg.IsInput()) {; |
| this.EncodeValue(enc, arg.Type, args[argIndex]) ; |
| argIndex += 1 ; |
| } |
| } |
| |
| Message msg = aBroker.CreateMessage(enc,obj.RoutingKey(),timeToLive) ; |
| |
| if (synchronous) { |
| aBroker.SetSyncInFlight(true) ; |
| } |
| aBroker.Send(msg) ; |
| } |
| return seq ; |
| } |
| |
| public QMFObject MakeObject(ClassKey key) { |
| SchemaClass sClass = this.GetSchema(key) ; |
| if (sClass == null) { |
| throw new Exception("No schema found for class " + key.ToString()) ; |
| } |
| |
| return this.CreateQMFObject(sClass, true, true, false) ; |
| } |
| |
| public QMFObject MakeObject(String keyString) { |
| return this.MakeObject(new ClassKey(keyString)) ; |
| } |
| |
| // Callback Methods |
| public void HandleNewAgent(Agent agent) { |
| if (Console != null) { |
| Console.NewAgent(agent) ; |
| } |
| } |
| |
| public void HandleAgentRemoved(Agent agent) { |
| if (Console != null) { |
| Console.AgentRemoved(agent) ; |
| } |
| } |
| |
| public void HandleBrokerConnect(Broker broker) { |
| if (Console != null) { |
| Console.BrokerConnected(broker) ; |
| } |
| } |
| |
| public void HandleBrokerDisconnect(Broker broker) { |
| if (Console != null) { |
| Console.BrokerDisconnected(broker) ; |
| } |
| } |
| |
| public void HandleBrokerResponse(Broker broker, IDecoder decoder, long sequence) { |
| if (Console != null) { |
| Console.BrokerInformation(broker) ; |
| } |
| |
| long seq = SequenceManager.Reserve(CONTEXT_STARTUP) ; |
| IEncoder endocder = broker.CreateEncoder('P', seq) ; |
| broker.Send(endocder) ; |
| } |
| |
| public void HandlePackageIndicator(Broker broker, IDecoder decoder, long sequence) { |
| string packageName = decoder.ReadStr8() ; |
| bool notify = false ; |
| if (!Packages.ContainsKey(packageName)) { |
| lock (LockObject) { |
| Packages[packageName] = new Dictionary<string, SchemaClass>() ; |
| notify = true ; |
| } |
| } |
| |
| if (notify && Console != null) { |
| Console.NewPackage(packageName) ; |
| } |
| |
| broker.IncrementOutstanding() ; |
| long seq = SequenceManager.Reserve(Session.CONTEXT_STARTUP) ; |
| IEncoder enc = broker.CreateEncoder('Q', seq) ; |
| enc.WriteStr8(packageName) ; |
| broker.Send(enc) ; |
| } |
| |
| public void HandleCommandComplete(Broker broker, IDecoder decoder, long sequence) { |
| |
| long code = decoder.ReadUint32() ; |
| string text = decoder.ReadStr8() ; |
| Object context = this.SequenceManager.Release(sequence) ; |
| |
| if (context.Equals(CONTEXT_STARTUP)) { |
| broker.DecrementOutstanding() ; |
| } else { |
| if ((context.Equals(CONTEXT_SYNC)) & broker.GetSyncInFlight()) { |
| broker.SetSyncInFlight(false) ; |
| } else { |
| if (context.Equals(CONTEXT_MULTIGET) && SyncSequenceList.Contains(sequence)) { |
| lock(LockObject) { |
| SyncSequenceList.Remove(sequence) ; |
| if (SyncSequenceList.Count == 0) { |
| Monitor.PulseAll(LockObject) ; |
| } |
| } |
| } |
| } |
| } |
| } |
| |
| public void HandleClassIndicator(Broker broker, IDecoder decoder, long sequence) { |
| short kind = decoder.ReadUint8() ; |
| ClassKey classKey = new ClassKey(decoder) ; |
| bool unknown = false ; |
| |
| |
| lock (LockObject) { |
| if (Packages.ContainsKey(classKey.PackageName)) { |
| if (!Packages[classKey.PackageName].ContainsKey(classKey.GetKeyString())) { |
| unknown = true ; |
| } |
| } |
| } |
| |
| if (unknown) { |
| broker.IncrementOutstanding() ; |
| long seq = SequenceManager.Reserve(Session.CONTEXT_STARTUP) ; |
| IEncoder enc = broker.CreateEncoder('S', seq) ; |
| classKey.encode(enc) ; |
| broker.Send(enc) ; |
| } |
| } |
| |
| public void HandleMethodResponse(Broker broker, IDecoder decoder, long sequence) { |
| long code = decoder.ReadUint32() ; |
| string text = decoder.ReadStr16() ; |
| |
| Dictionary<string, object> outArgs = new Dictionary<string, object>() ; |
| object obj = SequenceManager.Release(sequence) ; |
| |
| if (obj == null) { |
| return ; |
| } |
| |
| KeyValuePair<SchemaMethod, bool> pair = (KeyValuePair<SchemaMethod, bool>) obj ; |
| if (code == 0) { |
| foreach (SchemaArgument arg in pair.Key.Arguments) { |
| if (arg.IsOutput()) { |
| outArgs.Add(arg.Name, this.DecodeValue(decoder, arg.Type)) ; |
| } |
| } |
| } |
| |
| MethodResult result = new MethodResult(code, text, outArgs) ; |
| |
| if (pair.Value) { |
| this.SyncResult = result; |
| broker.SetSyncInFlight(false) ; |
| } |
| |
| if (Console != null) { |
| Console.MethodResponse(broker, sequence, result) ; |
| } |
| } |
| |
| public void HandleHeartbeatIndicator(Broker broker, IDecoder decoder, long sequence, IMessage msg) { |
| if (Console != null) { |
| long brokerBank = 1 ; |
| long agentBank = 0 ; |
| try { |
| string routingKey = msg.DeliveryProperties.GetRoutingKey() ; |
| if (routingKey != null) { |
| agentBank = Agent.GetBrokerBank(routingKey) ; |
| brokerBank = Agent.GetBrokerBank(routingKey) ; |
| } |
| } |
| catch (Exception e) { |
| log.Warn("Internal QPID error", e) ; |
| } |
| |
| string agentKey = Agent.AgentKey(agentBank, brokerBank) ; |
| long timestamp = decoder.ReadUint64() ; |
| if (broker.Agents.ContainsKey(agentKey)) { |
| Agent agent = broker.Agents[agentKey] ; |
| Console.HearbeatRecieved(agent, timestamp) ; |
| } |
| } |
| |
| } |
| |
| public void HandleEventIndicator(Broker broker, IDecoder decoder, long sequence) { |
| if (Console != null) { |
| QMFEvent newEvent = new QMFEvent(this, decoder) ; |
| Console.EventRecieved(broker, newEvent) ; |
| } |
| } |
| |
| public void HandleSchemaResponse(Broker broker, IDecoder decoder, long sequence) { |
| short kind = decoder.ReadUint8() ; |
| ClassKey classKey = new ClassKey(decoder) ; |
| SchemaClass sClass = new SchemaClass(kind, classKey, decoder, this) ; |
| lock(LockObject) { |
| Dictionary<string, SchemaClass> classMappings = Packages[sClass.PackageName] ; |
| classMappings.Remove(sClass.ClassKeyString) ; |
| classMappings.Add(sClass.ClassKeyString, sClass) ; |
| } |
| |
| SequenceManager.Release(sequence) ; |
| broker.DecrementOutstanding() ; |
| if (Console != null) { |
| this.Console.NewClass(kind, classKey) ; |
| } |
| } |
| |
| public void HandleContentIndicator(Broker broker, IDecoder decoder, long sequence, bool hasProperties, bool hasStatistics) { |
| |
| ClassKey key = new ClassKey(decoder) ; |
| SchemaClass sClass = null ;; |
| lock (LockObject) { |
| sClass = GetSchema(key, false) ; |
| } |
| if (sClass != null) { |
| QMFObject obj = this.CreateQMFObject(sClass, decoder, hasProperties, hasStatistics, true) ; |
| |
| if (key.PackageName.Equals("org.apache.qpid.broker") && key.ClassName.Equals("agent") && hasProperties) { |
| broker.UpdateAgent(obj) ; |
| } |
| |
| lock (LockObject) { |
| if (SyncSequenceList.Contains(sequence)) { |
| if (!obj.IsDeleted() && this.SelectMatch(obj)) { |
| GetResult.Add(obj) ; |
| } |
| } |
| } |
| |
| if (Console != null) { |
| if (hasProperties) { |
| Console.ObjectProperties(broker, obj) ; |
| } |
| if (hasStatistics) { |
| Console.ObjectStatistics(broker, obj) ; |
| } |
| } |
| } |
| } |
| |
| public bool SelectMatch(QMFObject obj) { |
| return true ; |
| } |
| |
| public object DecodeValue(IDecoder dec, short type) { |
| |
| switch (type) { |
| case 1: return dec.ReadUint8() ; // U8 |
| case 2: return dec.ReadUint16() ; // U16 |
| case 3: return dec.ReadUint32() ; // U32 |
| case 4: return dec.ReadUint64() ; // U64 |
| case 6: return dec.ReadStr8() ; // SSTR |
| case 7: return dec.ReadStr16() ; // LSTR |
| case 8: return dec.ReadDatetime() ; // ABSTIME |
| case 9: return dec.ReadUint32() ; // DELTATIME |
| case 10: return new ObjectID(dec) ; // ref |
| case 11: return dec.ReadUint8() != 0 ; // bool |
| case 12: return dec.ReadFloat() ; // float |
| case 13: return dec.ReadDouble() ; // double |
| case 14: return dec.ReadUuid() ; // UUID |
| case 15: return dec.ReadMap() ; // Ftable |
| case 16: return dec.ReadInt8() ; // int8 |
| case 17: return dec.ReadInt16() ; // int16 |
| case 18: return dec.ReadInt32() ; // int32 |
| case 19: return dec.ReadInt64() ; // int64 |
| case 20: // Object |
| // Peek into the inner type code, make sure |
| // it is actually an object |
| object returnValue = null ; |
| short innerTypeCode = dec.ReadUint8() ; |
| if (innerTypeCode != 20) { |
| returnValue = this.DecodeValue(dec, innerTypeCode) ; |
| } |
| else { |
| ClassKey classKey = new ClassKey(dec) ; |
| lock(LockObject) { |
| SchemaClass sClass = GetSchema(classKey) ; |
| if (sClass != null) { |
| returnValue = this.CreateQMFObject(sClass, dec, true, true, false) ; |
| } |
| } |
| } |
| return returnValue; |
| case 21: // List |
| { |
| MSDecoder lDec = new MSDecoder(); |
| lDec.Init(new MemoryStream(dec.ReadVbin32())); |
| long count = lDec.ReadUint32(); |
| List<object> newList = new List<object>(); |
| while (count > 0) |
| { |
| short innerType = lDec.ReadUint8(); |
| newList.Add(this.DecodeValue(lDec, innerType)); |
| count -= 1; |
| } |
| return newList; |
| } |
| case 22: // Array |
| { |
| MSDecoder aDec = new MSDecoder(); |
| aDec.Init(new MemoryStream(dec.ReadVbin32())); |
| long cnt = aDec.ReadUint32(); |
| short innerType = aDec.ReadUint8(); |
| List<object> aList = new List<object>(); |
| while (cnt > 0) |
| { |
| aList.Add(this.DecodeValue(aDec, innerType)); |
| cnt -= 1; |
| } |
| return aList; |
| } |
| default: |
| throw new Exception(String.Format("Invalid Type Code: {0}", type)) ; |
| } |
| } |
| |
| |
| public void EncodeValue(IEncoder enc, short type, object val) { |
| try { |
| switch ((int)type) { |
| case 1: enc.WriteUint8((short) val) ; break; // U8 |
| case 2: enc.WriteUint16((int) val) ; break; // U16 |
| case 3: enc.WriteUint32((long) val) ; break; // U32 |
| case 4: enc.WriteUint64((long) val) ; break; // U64 |
| case 6: enc.WriteStr8((string) val) ; break; // SSTR |
| case 7: enc.WriteStr16((string) val) ; break; // LSTR |
| case 8: enc.WriteDatetime((long) val); break; // ABSTIME |
| case 9: enc.WriteUint32((long) val); break; // DELTATIME |
| case 10: ((ObjectID)val).encode(enc) ; break; // ref |
| case 11: |
| if ((bool) val) { |
| enc.WriteUint8(1) ; |
| } else { |
| enc.WriteUint8(0) ; |
| } |
| break ; |
| case 12: enc.WriteFloat((float) val); break; // FLOAT |
| case 13: enc.WriteDouble((double) val); break; // DOUBLE |
| case 14: enc.WriteUuid((UUID) val) ; break ; // UUID |
| case 15: enc.WriteMap((Dictionary<string, object>) val) ; break ; // Ftable |
| case 16: enc.WriteInt8((short) val) ; break; // int8 |
| case 17: enc.WriteInt16((int) val) ; break; // int16 |
| case 18: enc.WriteInt32(long.Parse(""+ val)) ; break; // int32 |
| case 19: enc.WriteInt64(long.Parse("" + val)) ; break; // int64 |
| case 20: // Object |
| // Check that the object has a session, if not |
| // take ownership of it |
| QMFObject qObj = (QMFObject) val ; |
| if (qObj.Session == null) { |
| qObj.Session = this ; |
| } |
| qObj.Encode(enc) ; |
| break; |
| case 21: // List |
| List<object> items = (List<object>) val ; |
| MSEncoder lEnc = new MSEncoder(1) ; |
| lEnc.Init() ; |
| lEnc.WriteUint32(items.Count) ; |
| foreach (object obj in items) { |
| short innerType = Util.QMFType(obj) ; |
| lEnc.WriteUint8(innerType) ; |
| this.EncodeValue(lEnc,innerType,obj) ; |
| } |
| enc.WriteVbin32(lEnc.Segment().ToArray()) ; |
| break ; |
| case 22: // Array |
| List<object> aItems = (List<object>) val ; |
| MSEncoder aEnc = new MSEncoder(1) ; |
| aEnc.Init() ; |
| long aCount = aItems.Count ; |
| aEnc.WriteUint32(aCount) ; |
| if (aCount > 0) { |
| Object anObj = aItems[0] ; |
| short innerType = Util.QMFType(anObj) ; |
| aEnc.WriteUint8(innerType) ; |
| foreach (object obj in aItems) { |
| this.EncodeValue(aEnc,innerType,obj) ; |
| } |
| } |
| enc.WriteVbin32(aEnc.Segment().ToArray()) ; |
| break ; |
| default: |
| throw new Exception(String.Format("Invalid Type Code: {0}", type)) ; |
| } |
| } |
| catch (System.InvalidCastException e) { |
| string msg = String.Format("Class cast exception for typecode {0}, type {1} ", type, val.GetType()) ; |
| log.Error(msg) ; |
| throw new Exception(msg + type, e) ; |
| } |
| } |
| |
| public List<string> BindingKeys() { |
| List<string> bindings = new List<string>() ; |
| bindings.Add("schema.#") ; |
| if (RecieveObjects & RecieveEvents & RecieveHeartbeat & !UserBindings) { |
| bindings.Add("console.#") ; |
| } |
| else { |
| if (RecieveObjects & !UserBindings) { |
| bindings.Add("console.obj.#") ; |
| } |
| else { |
| bindings.Add("console.obj.*.*.org.apache.qpid.broker.agent") ; |
| } |
| if (RecieveEvents) { |
| bindings.Add("console.event.#") ; |
| } |
| if (RecieveHeartbeat) { |
| bindings.Add("console.heartbeat.#") ; |
| } |
| } |
| return bindings ; |
| } |
| |
| protected QMFObject CreateQMFObject(SchemaClass schema, bool hasProperties, bool hasStats , bool isManaged) { |
| Type realClass = typeof(QMFObject) ; |
| if (Console != null) { |
| realClass = Console.TypeMapping(schema.Key) ; |
| } |
| Type[] types = new Type[] {typeof(Session), typeof(SchemaClass), typeof(bool), typeof(bool),typeof(bool)} ; |
| object[] args = new object[] {this, schema, hasProperties, hasStats, isManaged} ; |
| ConstructorInfo ci = realClass.GetConstructor(types); |
| return (QMFObject) ci.Invoke(args) ; |
| } |
| |
| protected QMFObject CreateQMFObject(SchemaClass schema, IDecoder dec, bool hasProperties, bool hasStats , bool isManaged) { |
| Type realClass = typeof(QMFObject) ; |
| if (Console != null) { |
| realClass = Console.TypeMapping(schema.Key) ; |
| } |
| Type[] types = new Type[] {typeof(Session), typeof(SchemaClass), typeof(IDecoder), typeof(bool), typeof(bool),typeof(bool)} ; |
| object[] args = new object[] {this, schema, dec, hasProperties, hasStats, isManaged} ; |
| ConstructorInfo ci = realClass.GetConstructor(types); |
| return (QMFObject) ci.Invoke(args) ; |
| } |
| } |
| } |
