diff options
Diffstat (limited to 'cpp/src/qmf/engine/BrokerProxyImpl.cpp')
| -rw-r--r-- | cpp/src/qmf/engine/BrokerProxyImpl.cpp | 827 |
1 files changed, 0 insertions, 827 deletions
diff --git a/cpp/src/qmf/engine/BrokerProxyImpl.cpp b/cpp/src/qmf/engine/BrokerProxyImpl.cpp deleted file mode 100644 index 5fc71979fd..0000000000 --- a/cpp/src/qmf/engine/BrokerProxyImpl.cpp +++ /dev/null @@ -1,827 +0,0 @@ -/* - * 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. - */ - -#include "qmf/engine/BrokerProxyImpl.h" -#include "qmf/engine/ConsoleImpl.h" -#include "qmf/engine/Protocol.h" -#include "qpid/Address.h" -#include "qpid/sys/SystemInfo.h" -#include <qpid/log/Statement.h> -#include <qpid/StringUtils.h> -#include <string.h> -#include <iostream> -#include <fstream> - -using namespace std; -using namespace qmf::engine; -using namespace qpid::framing; -using namespace qpid::sys; - -namespace { - const char* QMF_EXCHANGE = "qpid.management"; - const char* DIR_EXCHANGE = "amq.direct"; - const char* BROKER_KEY = "broker"; - const char* BROKER_PACKAGE = "org.apache.qpid.broker"; - const char* AGENT_CLASS = "agent"; - const char* BROKER_AGENT_KEY = "agent.1.0"; -} - -const Object* QueryResponseImpl::getObject(uint32_t idx) const -{ - vector<ObjectPtr>::const_iterator iter = results.begin(); - - while (idx > 0) { - if (iter == results.end()) - return 0; - iter++; - idx--; - } - - return iter->get(); -} - -#define STRING_REF(s) {if (!s.empty()) item.s = const_cast<char*>(s.c_str());} - -BrokerEvent BrokerEventImpl::copy() -{ - BrokerEvent item; - - ::memset(&item, 0, sizeof(BrokerEvent)); - item.kind = kind; - - STRING_REF(name); - STRING_REF(exchange); - STRING_REF(bindingKey); - item.context = context; - item.queryResponse = queryResponse.get(); - item.methodResponse = methodResponse.get(); - - return item; -} - -BrokerProxyImpl::BrokerProxyImpl(BrokerProxy& pub, Console& _console) : publicObject(pub), console(_console) -{ - stringstream qn; - qpid::Address addr; - - SystemInfo::getLocalHostname(addr); - qn << "qmfc-" << SystemInfo::getProcessName() << "-" << addr << "-" << SystemInfo::getProcessId(); - queueName = qn.str(); - - seqMgr.setUnsolicitedContext(SequenceContext::Ptr(new StaticContext(*this))); -} - -void BrokerProxyImpl::sessionOpened(SessionHandle& /*sh*/) -{ - Mutex::ScopedLock _lock(lock); - agentList.clear(); - eventQueue.clear(); - xmtQueue.clear(); - eventQueue.push_back(eventDeclareQueue(queueName)); - eventQueue.push_back(eventBind(DIR_EXCHANGE, queueName, queueName)); - eventQueue.push_back(eventSetupComplete()); - - // TODO: Store session handle -} - -void BrokerProxyImpl::sessionClosed() -{ - Mutex::ScopedLock _lock(lock); - agentList.clear(); - eventQueue.clear(); - xmtQueue.clear(); -} - -void BrokerProxyImpl::startProtocol() -{ - AgentProxyPtr agent(AgentProxyImpl::factory(console, publicObject, 0, "Agent embedded in broker")); - { - Mutex::ScopedLock _lock(lock); - char rawbuffer[512]; - Buffer buffer(rawbuffer, 512); - - agentList[0] = agent; - - requestsOutstanding = 1; - topicBound = false; - uint32_t sequence(seqMgr.reserve()); - Protocol::encodeHeader(buffer, Protocol::OP_BROKER_REQUEST, sequence); - sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); - QPID_LOG(trace, "SENT BrokerRequest seq=" << sequence); - } - - console.impl->eventAgentAdded(agent); -} - -void BrokerProxyImpl::sendBufferLH(Buffer& buf, const string& destination, const string& routingKey) -{ - uint32_t length = buf.getPosition(); - MessageImpl::Ptr message(new MessageImpl); - - buf.reset(); - buf.getRawData(message->body, length); - message->destination = destination; - message->routingKey = routingKey; - message->replyExchange = DIR_EXCHANGE; - message->replyKey = queueName; - - xmtQueue.push_back(message); -} - -void BrokerProxyImpl::handleRcvMessage(Message& message) -{ - Buffer inBuffer(message.body, message.length); - uint8_t opcode; - uint32_t sequence; - - while (Protocol::checkHeader(inBuffer, &opcode, &sequence)) - seqMgr.dispatch(opcode, sequence, message.routingKey ? string(message.routingKey) : string(), inBuffer); -} - -bool BrokerProxyImpl::getXmtMessage(Message& item) const -{ - Mutex::ScopedLock _lock(lock); - if (xmtQueue.empty()) - return false; - item = xmtQueue.front()->copy(); - return true; -} - -void BrokerProxyImpl::popXmt() -{ - Mutex::ScopedLock _lock(lock); - if (!xmtQueue.empty()) - xmtQueue.pop_front(); -} - -bool BrokerProxyImpl::getEvent(BrokerEvent& event) const -{ - Mutex::ScopedLock _lock(lock); - if (eventQueue.empty()) - return false; - event = eventQueue.front()->copy(); - return true; -} - -void BrokerProxyImpl::popEvent() -{ - Mutex::ScopedLock _lock(lock); - if (!eventQueue.empty()) - eventQueue.pop_front(); -} - -uint32_t BrokerProxyImpl::agentCount() const -{ - Mutex::ScopedLock _lock(lock); - return agentList.size(); -} - -const AgentProxy* BrokerProxyImpl::getAgent(uint32_t idx) const -{ - Mutex::ScopedLock _lock(lock); - for (map<uint32_t, AgentProxyPtr>::const_iterator iter = agentList.begin(); - iter != agentList.end(); iter++) - if (idx-- == 0) - return iter->second.get(); - return 0; -} - -void BrokerProxyImpl::sendQuery(const Query& query, void* context, const AgentProxy* agent) -{ - SequenceContext::Ptr queryContext(new QueryContext(*this, context)); - Mutex::ScopedLock _lock(lock); - bool sent = false; - if (agent != 0) { - if (sendGetRequestLH(queryContext, query, agent)) - sent = true; - } else { - // TODO (optimization) only send queries to agents that have the requested class+package - for (map<uint32_t, AgentProxyPtr>::const_iterator iter = agentList.begin(); - iter != agentList.end(); iter++) { - if (sendGetRequestLH(queryContext, query, iter->second.get())) - sent = true; - } - } - - if (!sent) { - queryContext->reserve(); - queryContext->release(); - } -} - -bool BrokerProxyImpl::sendGetRequestLH(SequenceContext::Ptr queryContext, const Query& query, const AgentProxy* agent) -{ - if (query.impl->singleAgent()) { - if (query.impl->agentBank() != agent->getAgentBank()) - return false; - } - stringstream key; - Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); - uint32_t sequence(seqMgr.reserve(queryContext)); - agent->impl->addSequence(sequence); - - Protocol::encodeHeader(outBuffer, Protocol::OP_GET_QUERY, sequence); - query.impl->encode(outBuffer); - key << "agent.1." << agent->impl->agentBank; - sendBufferLH(outBuffer, QMF_EXCHANGE, key.str()); - QPID_LOG(trace, "SENT GetQuery seq=" << sequence << " key=" << key.str()); - return true; -} - -string BrokerProxyImpl::encodeMethodArguments(const SchemaMethod* schema, const Value* argmap, Buffer& buffer) -{ - int argCount = schema->getArgumentCount(); - - if (argmap == 0 || !argmap->isMap()) - return string("Arguments must be in a map value"); - - for (int aIdx = 0; aIdx < argCount; aIdx++) { - const SchemaArgument* arg(schema->getArgument(aIdx)); - if (arg->getDirection() == DIR_IN || arg->getDirection() == DIR_IN_OUT) { - if (argmap->keyInMap(arg->getName())) { - const Value* argVal(argmap->byKey(arg->getName())); - if (argVal->getType() != arg->getType()) - return string("Argument is the wrong type: ") + arg->getName(); - argVal->impl->encode(buffer); - } else { - Value defaultValue(arg->getType()); - defaultValue.impl->encode(buffer); - } - } - } - - return string(); -} - -string BrokerProxyImpl::encodedSizeMethodArguments(const SchemaMethod* schema, const Value* argmap, uint32_t& size) -{ - int argCount = schema->getArgumentCount(); - - if (argmap == 0 || !argmap->isMap()) - return string("Arguments must be in a map value"); - - for (int aIdx = 0; aIdx < argCount; aIdx++) { - const SchemaArgument* arg(schema->getArgument(aIdx)); - if (arg->getDirection() == DIR_IN || arg->getDirection() == DIR_IN_OUT) { - if (argmap->keyInMap(arg->getName())) { - const Value* argVal(argmap->byKey(arg->getName())); - if (argVal->getType() != arg->getType()) - return string("Argument is the wrong type: ") + arg->getName(); - size += argVal->impl->encodedSize(); - } else { - Value defaultValue(arg->getType()); - size += defaultValue.impl->encodedSize(); - } - } - } - - return string(); -} - -void BrokerProxyImpl::sendMethodRequest(ObjectId* oid, const SchemaObjectClass* cls, - const string& methodName, const Value* args, void* userContext) -{ - int methodCount = cls->getMethodCount(); - int idx; - for (idx = 0; idx < methodCount; idx++) { - const SchemaMethod* method = cls->getMethod(idx); - if (string(method->getName()) == methodName) { - Mutex::ScopedLock _lock(lock); - SequenceContext::Ptr methodContext(new MethodContext(*this, userContext, method)); - stringstream key; - char* buf(outputBuffer); - uint32_t bufLen(1024); - bool allocated(false); - - string argErrorString = encodedSizeMethodArguments(method, args, bufLen); - if (!argErrorString.empty()) { - MethodResponsePtr argError(MethodResponseImpl::factory(1, argErrorString)); - eventQueue.push_back(eventMethodResponse(userContext, argError)); - return; - } - - if (bufLen > MA_BUFFER_SIZE) { - buf = (char*) malloc(bufLen); - allocated = true; - } - - Buffer outBuffer(buf, bufLen); - uint32_t sequence(seqMgr.reserve(methodContext)); - - Protocol::encodeHeader(outBuffer, Protocol::OP_METHOD_REQUEST, sequence); - oid->impl->encode(outBuffer); - cls->getClassKey()->impl->encode(outBuffer); - outBuffer.putShortString(methodName); - - encodeMethodArguments(method, args, outBuffer); - key << "agent.1." << oid->impl->getAgentBank(); - sendBufferLH(outBuffer, QMF_EXCHANGE, key.str()); - QPID_LOG(trace, "SENT MethodRequest seq=" << sequence << " method=" << methodName << " key=" << key.str()); - - if (allocated) - free(buf); - - return; - } - } - - MethodResponsePtr error(MethodResponseImpl::factory(1, string("Unknown method: ") + methodName)); - Mutex::ScopedLock _lock(lock); - eventQueue.push_back(eventMethodResponse(userContext, error)); -} - -void BrokerProxyImpl::addBinding(const string& exchange, const string& key) -{ - Mutex::ScopedLock _lock(lock); - eventQueue.push_back(eventBind(exchange, queueName, key)); -} - -BrokerEventImpl::Ptr BrokerProxyImpl::eventDeclareQueue(const string& queueName) -{ - BrokerEventImpl::Ptr event(new BrokerEventImpl(BrokerEvent::DECLARE_QUEUE)); - event->name = queueName; - return event; -} - -BrokerEventImpl::Ptr BrokerProxyImpl::eventBind(const string& exchange, const string& queue, const string& key) -{ - BrokerEventImpl::Ptr event(new BrokerEventImpl(BrokerEvent::BIND)); - event->name = queue; - event->exchange = exchange; - event->bindingKey = key; - - return event; -} - -BrokerEventImpl::Ptr BrokerProxyImpl::eventSetupComplete() -{ - BrokerEventImpl::Ptr event(new BrokerEventImpl(BrokerEvent::SETUP_COMPLETE)); - return event; -} - -BrokerEventImpl::Ptr BrokerProxyImpl::eventStable() -{ - QPID_LOG(trace, "Console Link to Broker Stable"); - BrokerEventImpl::Ptr event(new BrokerEventImpl(BrokerEvent::STABLE)); - return event; -} - -BrokerEventImpl::Ptr BrokerProxyImpl::eventQueryComplete(void* context, QueryResponsePtr response) -{ - BrokerEventImpl::Ptr event(new BrokerEventImpl(BrokerEvent::QUERY_COMPLETE)); - event->context = context; - event->queryResponse = response; - return event; -} - -BrokerEventImpl::Ptr BrokerProxyImpl::eventMethodResponse(void* context, MethodResponsePtr response) -{ - BrokerEventImpl::Ptr event(new BrokerEventImpl(BrokerEvent::METHOD_RESPONSE)); - event->context = context; - event->methodResponse = response; - return event; -} - -void BrokerProxyImpl::handleBrokerResponse(Buffer& inBuffer, uint32_t seq) -{ - brokerId.decode(inBuffer); - QPID_LOG(trace, "RCVD BrokerResponse seq=" << seq << " brokerId=" << brokerId); - Mutex::ScopedLock _lock(lock); - Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); - uint32_t sequence(seqMgr.reserve()); - incOutstandingLH(); - Protocol::encodeHeader(outBuffer, Protocol::OP_PACKAGE_REQUEST, sequence); - sendBufferLH(outBuffer, QMF_EXCHANGE, BROKER_KEY); - QPID_LOG(trace, "SENT PackageRequest seq=" << sequence); -} - -void BrokerProxyImpl::handlePackageIndication(Buffer& inBuffer, uint32_t seq) -{ - string package; - - inBuffer.getShortString(package); - QPID_LOG(trace, "RCVD PackageIndication seq=" << seq << " package=" << package); - console.impl->learnPackage(package); - - Mutex::ScopedLock _lock(lock); - Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); - uint32_t sequence(seqMgr.reserve()); - incOutstandingLH(); - Protocol::encodeHeader(outBuffer, Protocol::OP_CLASS_QUERY, sequence); - outBuffer.putShortString(package); - sendBufferLH(outBuffer, QMF_EXCHANGE, BROKER_KEY); - QPID_LOG(trace, "SENT ClassQuery seq=" << sequence << " package=" << package); -} - -void BrokerProxyImpl::handleCommandComplete(Buffer& inBuffer, uint32_t seq) -{ - string text; - uint32_t code = inBuffer.getLong(); - inBuffer.getShortString(text); - QPID_LOG(trace, "RCVD CommandComplete seq=" << seq << " code=" << code << " text=" << text); -} - -void BrokerProxyImpl::handleClassIndication(Buffer& inBuffer, uint32_t seq) -{ - uint8_t kind = inBuffer.getOctet(); - auto_ptr<SchemaClassKey> classKey(SchemaClassKeyImpl::factory(inBuffer)); - - QPID_LOG(trace, "RCVD ClassIndication seq=" << seq << " kind=" << (int) kind << " key=" << classKey->impl->str()); - - if (!console.impl->haveClass(classKey.get())) { - Mutex::ScopedLock _lock(lock); - incOutstandingLH(); - Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); - uint32_t sequence(seqMgr.reserve()); - Protocol::encodeHeader(outBuffer, Protocol::OP_SCHEMA_REQUEST, sequence); - classKey->impl->encode(outBuffer); - sendBufferLH(outBuffer, QMF_EXCHANGE, BROKER_KEY); - QPID_LOG(trace, "SENT SchemaRequest seq=" << sequence <<" key=" << classKey->impl->str()); - } -} - -MethodResponsePtr BrokerProxyImpl::handleMethodResponse(Buffer& inBuffer, uint32_t seq, const SchemaMethod* schema) -{ - MethodResponsePtr response(MethodResponseImpl::factory(inBuffer, schema)); - - QPID_LOG(trace, "RCVD MethodResponse seq=" << seq << " status=" << response->getStatus() << " text=" << - response->getException()->asString()); - - return response; -} - -void BrokerProxyImpl::handleHeartbeatIndication(Buffer& inBuffer, uint32_t seq, const string& routingKey) -{ - vector<string> tokens = qpid::split(routingKey, "."); - uint32_t agentBank; - uint64_t timestamp; - - if (routingKey.empty() || tokens.size() != 4) - agentBank = 0; - else - agentBank = ::atoi(tokens[3].c_str()); - - timestamp = inBuffer.getLongLong(); - map<uint32_t, AgentProxyPtr>::const_iterator iter = agentList.find(agentBank); - if (iter != agentList.end()) { - console.impl->eventAgentHeartbeat(iter->second, timestamp); - } - QPID_LOG(trace, "RCVD HeartbeatIndication seq=" << seq << " agentBank=" << agentBank); -} - -void BrokerProxyImpl::handleEventIndication(Buffer& inBuffer, uint32_t seq) -{ - auto_ptr<SchemaClassKey> classKey(SchemaClassKeyImpl::factory(inBuffer)); - const SchemaEventClass *schema = console.impl->getEventClass(classKey.get()); - if (schema == 0) { - QPID_LOG(trace, "No Schema Found for EventIndication. seq=" << seq << " key=" << classKey->impl->str()); - return; - } - - EventPtr eptr(EventImpl::factory(schema, inBuffer)); - - console.impl->eventEventReceived(eptr); - QPID_LOG(trace, "RCVD EventIndication seq=" << seq << " key=" << classKey->impl->str()); -} - -void BrokerProxyImpl::handleSchemaResponse(Buffer& inBuffer, uint32_t seq) -{ - SchemaObjectClass* oClassPtr; - SchemaEventClass* eClassPtr; - uint8_t kind = inBuffer.getOctet(); - const SchemaClassKey* key; - if (kind == CLASS_OBJECT) { - oClassPtr = SchemaObjectClassImpl::factory(inBuffer); - console.impl->learnClass(oClassPtr); - key = oClassPtr->getClassKey(); - QPID_LOG(trace, "RCVD SchemaResponse seq=" << seq << " kind=object key=" << key->impl->str()); - - // - // If we have just learned about the org.apache.qpid.broker:agent class, send a get - // request for the current list of agents so we can have it on-hand before we declare - // this session "stable". - // - if (key->impl->getClassName() == AGENT_CLASS && key->impl->getPackageName() == BROKER_PACKAGE) { - Mutex::ScopedLock _lock(lock); - incOutstandingLH(); - Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE); - uint32_t sequence(seqMgr.reserve()); - Protocol::encodeHeader(outBuffer, Protocol::OP_GET_QUERY, sequence); - FieldTable ft; - ft.setString("_class", AGENT_CLASS); - ft.setString("_package", BROKER_PACKAGE); - ft.encode(outBuffer); - sendBufferLH(outBuffer, QMF_EXCHANGE, BROKER_AGENT_KEY); - QPID_LOG(trace, "SENT GetQuery seq=" << sequence << " key=" << BROKER_AGENT_KEY); - } - } else if (kind == CLASS_EVENT) { - eClassPtr = SchemaEventClassImpl::factory(inBuffer); - console.impl->learnClass(eClassPtr); - key = eClassPtr->getClassKey(); - QPID_LOG(trace, "RCVD SchemaResponse seq=" << seq << " kind=event key=" << key->impl->str()); - } - else { - QPID_LOG(error, "BrokerProxyImpl::handleSchemaResponse received unknown class kind: " << (int) kind); - } -} - -ObjectPtr BrokerProxyImpl::handleObjectIndication(Buffer& inBuffer, uint32_t seq, bool prop, bool stat) -{ - auto_ptr<SchemaClassKey> classKey(SchemaClassKeyImpl::factory(inBuffer)); - QPID_LOG(trace, "RCVD ObjectIndication seq=" << seq << " key=" << classKey->impl->str()); - - SchemaObjectClass* schema = console.impl->getSchema(classKey.get()); - if (schema == 0) { - QPID_LOG(trace, "No Schema Found for ObjectIndication. seq=" << seq << " key=" << classKey->impl->str()); - return ObjectPtr(); - } - - ObjectPtr optr(ObjectImpl::factory(schema, this, inBuffer, prop, stat, true)); - if (prop && classKey->impl->getPackageName() == BROKER_PACKAGE && classKey->impl->getClassName() == AGENT_CLASS) { - // - // We've intercepted information about a remote agent... update the agent list accordingly - // - updateAgentList(optr); - } - return optr; -} - -void BrokerProxyImpl::updateAgentList(ObjectPtr obj) -{ - Value* value = obj->getValue("agentBank"); - Mutex::ScopedLock _lock(lock); - if (value != 0 && value->isUint()) { - uint32_t agentBank = value->asUint(); - if (obj->isDeleted()) { - map<uint32_t, AgentProxyPtr>::iterator iter = agentList.find(agentBank); - if (iter != agentList.end()) { - AgentProxyPtr agent(iter->second); - console.impl->eventAgentDeleted(agent); - agentList.erase(agentBank); - QPID_LOG(trace, "Agent at bank " << agentBank << " removed from agent list"); - - // - // Release all sequence numbers for requests in-flight to this agent. - // Since the agent is no longer connected, these requests would not - // otherwise complete. - // - agent->impl->releaseInFlight(seqMgr); - } - } else { - Value* str = obj->getValue("label"); - string label; - if (str != 0 && str->isString()) - label = str->asString(); - map<uint32_t, AgentProxyPtr>::const_iterator iter = agentList.find(agentBank); - if (iter == agentList.end()) { - AgentProxyPtr agent(AgentProxyImpl::factory(console, publicObject, agentBank, label)); - agentList[agentBank] = agent; - console.impl->eventAgentAdded(agent); - QPID_LOG(trace, "Agent '" << label << "' found at bank " << agentBank); - } - } - } -} - -void BrokerProxyImpl::incOutstandingLH() -{ - requestsOutstanding++; -} - -void BrokerProxyImpl::decOutstanding() -{ - Mutex::ScopedLock _lock(lock); - requestsOutstanding--; - if (requestsOutstanding == 0 && !topicBound) { - topicBound = true; - for (vector<pair<string, string> >::const_iterator iter = console.impl->bindingList.begin(); - iter != console.impl->bindingList.end(); iter++) { - string exchange(iter->first.empty() ? QMF_EXCHANGE : iter->first); - string key(iter->second); - eventQueue.push_back(eventBind(exchange, queueName, key)); - } - eventQueue.push_back(eventStable()); - } -} - -MethodResponseImpl::MethodResponseImpl(const MethodResponseImpl& from) : - status(from.status), schema(from.schema) -{ - if (from.exception.get()) - exception.reset(new Value(*(from.exception))); - if (from.arguments.get()) - arguments.reset(new Value(*(from.arguments))); -} - -MethodResponseImpl::MethodResponseImpl(Buffer& buf, const SchemaMethod* s) : schema(s) -{ - string text; - - status = buf.getLong(); - buf.getMediumString(text); - exception.reset(new Value(TYPE_LSTR)); - exception->setString(text.c_str()); - - if (status != 0) - return; - - arguments.reset(new Value(TYPE_MAP)); - int argCount(schema->getArgumentCount()); - for (int idx = 0; idx < argCount; idx++) { - const SchemaArgument* arg = schema->getArgument(idx); - if (arg->getDirection() == DIR_OUT || arg->getDirection() == DIR_IN_OUT) { - Value* value(ValueImpl::factory(arg->getType(), buf)); - arguments->insert(arg->getName(), value); - } - } -} - -MethodResponseImpl::MethodResponseImpl(uint32_t s, const string& text) : schema(0) -{ - status = s; - exception.reset(new Value(TYPE_LSTR)); - exception->setString(text.c_str()); -} - -MethodResponse* MethodResponseImpl::factory(Buffer& buf, const SchemaMethod* schema) -{ - MethodResponseImpl* impl(new MethodResponseImpl(buf, schema)); - return new MethodResponse(impl); -} - -MethodResponse* MethodResponseImpl::factory(uint32_t status, const std::string& text) -{ - MethodResponseImpl* impl(new MethodResponseImpl(status, text)); - return new MethodResponse(impl); -} - -bool StaticContext::handleMessage(uint8_t opcode, uint32_t sequence, const string& routingKey, Buffer& buffer) -{ - ObjectPtr object; - bool completeContext = false; - - if (opcode == Protocol::OP_BROKER_RESPONSE) { - broker.handleBrokerResponse(buffer, sequence); - completeContext = true; - } - else if (opcode == Protocol::OP_COMMAND_COMPLETE) { - broker.handleCommandComplete(buffer, sequence); - completeContext = true; - } - else if (opcode == Protocol::OP_SCHEMA_RESPONSE) { - broker.handleSchemaResponse(buffer, sequence); - completeContext = true; - } - else if (opcode == Protocol::OP_PACKAGE_INDICATION) - broker.handlePackageIndication(buffer, sequence); - else if (opcode == Protocol::OP_CLASS_INDICATION) - broker.handleClassIndication(buffer, sequence); - else if (opcode == Protocol::OP_HEARTBEAT_INDICATION) - broker.handleHeartbeatIndication(buffer, sequence, routingKey); - else if (opcode == Protocol::OP_EVENT_INDICATION) - broker.handleEventIndication(buffer, sequence); - else if (opcode == Protocol::OP_PROPERTY_INDICATION) { - object = broker.handleObjectIndication(buffer, sequence, true, false); - broker.console.impl->eventObjectUpdate(object, true, false); - } - else if (opcode == Protocol::OP_STATISTIC_INDICATION) { - object = broker.handleObjectIndication(buffer, sequence, false, true); - broker.console.impl->eventObjectUpdate(object, false, true); - } - else if (opcode == Protocol::OP_OBJECT_INDICATION) { - object = broker.handleObjectIndication(buffer, sequence, true, true); - broker.console.impl->eventObjectUpdate(object, true, true); - } - else { - QPID_LOG(trace, "StaticContext::handleMessage invalid opcode: " << opcode); - completeContext = true; - } - - return completeContext; -} - -void QueryContext::reserve() -{ - Mutex::ScopedLock _lock(lock); - requestsOutstanding++; -} - -void QueryContext::release() -{ - { - Mutex::ScopedLock _lock(lock); - if (--requestsOutstanding > 0) - return; - } - - Mutex::ScopedLock _block(broker.lock); - broker.eventQueue.push_back(broker.eventQueryComplete(userContext, queryResponse)); -} - -bool QueryContext::handleMessage(uint8_t opcode, uint32_t sequence, const string& /*routingKey*/, Buffer& buffer) -{ - bool completeContext = false; - ObjectPtr object; - - if (opcode == Protocol::OP_COMMAND_COMPLETE) { - broker.handleCommandComplete(buffer, sequence); - completeContext = true; - - // - // Visit each agent and remove the sequence from that agent's in-flight list. - // This could be made more efficient because only one agent will have this sequence - // in its list. - // - map<uint32_t, AgentProxyPtr> copy; - { - Mutex::ScopedLock _block(broker.lock); - copy = broker.agentList; - } - for (map<uint32_t, AgentProxyPtr>::iterator iter = copy.begin(); iter != copy.end(); iter++) - iter->second->impl->delSequence(sequence); - } - else if (opcode == Protocol::OP_OBJECT_INDICATION) { - object = broker.handleObjectIndication(buffer, sequence, true, true); - if (object.get() != 0) - queryResponse->impl->results.push_back(object); - } - else { - QPID_LOG(trace, "QueryContext::handleMessage invalid opcode: " << opcode); - completeContext = true; - } - - return completeContext; -} - -void MethodContext::release() -{ - Mutex::ScopedLock _block(broker.lock); - broker.eventQueue.push_back(broker.eventMethodResponse(userContext, methodResponse)); -} - -bool MethodContext::handleMessage(uint8_t opcode, uint32_t sequence, const string& /*routingKey*/, Buffer& buffer) -{ - if (opcode == Protocol::OP_METHOD_RESPONSE) - methodResponse = broker.handleMethodResponse(buffer, sequence, schema); - else - QPID_LOG(trace, "QueryContext::handleMessage invalid opcode: " << opcode); - - return true; -} - - -//================================================================== -// Wrappers -//================================================================== - -AgentProxy::AgentProxy(AgentProxyImpl* i) : impl(i) {} -AgentProxy::AgentProxy(const AgentProxy& from) : impl(new AgentProxyImpl(*(from.impl))) {} -AgentProxy::~AgentProxy() { delete impl; } -const char* AgentProxy::getLabel() const { return impl->getLabel().c_str(); } -uint32_t AgentProxy::getBrokerBank() const { return impl->getBrokerBank(); } -uint32_t AgentProxy::getAgentBank() const { return impl->getAgentBank(); } - -BrokerProxy::BrokerProxy(Console& console) : impl(new BrokerProxyImpl(*this, console)) {} -BrokerProxy::~BrokerProxy() { delete impl; } -void BrokerProxy::sessionOpened(SessionHandle& sh) { impl->sessionOpened(sh); } -void BrokerProxy::sessionClosed() { impl->sessionClosed(); } -void BrokerProxy::startProtocol() { impl->startProtocol(); } -void BrokerProxy::handleRcvMessage(Message& message) { impl->handleRcvMessage(message); } -bool BrokerProxy::getXmtMessage(Message& item) const { return impl->getXmtMessage(item); } -void BrokerProxy::popXmt() { impl->popXmt(); } -bool BrokerProxy::getEvent(BrokerEvent& event) const { return impl->getEvent(event); } -void BrokerProxy::popEvent() { impl->popEvent(); } -uint32_t BrokerProxy::agentCount() const { return impl->agentCount(); } -const AgentProxy* BrokerProxy::getAgent(uint32_t idx) const { return impl->getAgent(idx); } -void BrokerProxy::sendQuery(const Query& query, void* context, const AgentProxy* agent) { impl->sendQuery(query, context, agent); } - -MethodResponse::MethodResponse(const MethodResponse& from) : impl(new MethodResponseImpl(*(from.impl))) {} -MethodResponse::MethodResponse(MethodResponseImpl* i) : impl(i) {} -MethodResponse::~MethodResponse() {} -uint32_t MethodResponse::getStatus() const { return impl->getStatus(); } -const Value* MethodResponse::getException() const { return impl->getException(); } -const Value* MethodResponse::getArgs() const { return impl->getArgs(); } - -QueryResponse::QueryResponse(QueryResponseImpl* i) : impl(i) {} -QueryResponse::~QueryResponse() {} -uint32_t QueryResponse::getStatus() const { return impl->getStatus(); } -const Value* QueryResponse::getException() const { return impl->getException(); } -uint32_t QueryResponse::getObjectCount() const { return impl->getObjectCount(); } -const Object* QueryResponse::getObject(uint32_t idx) const { return impl->getObject(idx); } - |