/* * * Copyright (c) 2006 The Apache Software Foundation * * Licensed 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 "FanOutExchange.h" #include "ExchangeBinding.h" #include using namespace qpid::broker; using namespace qpid::framing; using namespace qpid::concurrent; FanOutExchange::FanOutExchange(const std::string& name) : Exchange(name) {} void FanOutExchange::bind(Queue::shared_ptr queue, const string& routingKey, FieldTable* args){ Locker locker(lock); // Add if not already present. Queue::vector::iterator i = std::find(bindings.begin(), bindings.end(), queue); if (i == bindings.end()) { bindings.push_back(queue); queue->bound(new ExchangeBinding(this, queue, routingKey, args)); } } void FanOutExchange::unbind(Queue::shared_ptr queue, const string& routingKey, FieldTable* args){ Locker locker(lock); Queue::vector::iterator i = std::find(bindings.begin(), bindings.end(), queue); if (i != bindings.end()) { bindings.erase(i); // TODO aconway 2006-09-14: What about the ExchangeBinding object? Don't we have to verify routingKey/args match? } } void FanOutExchange::route(Message::shared_ptr& msg, const string& routingKey, FieldTable* args){ Locker locker(lock); for(Queue::vector::iterator i = bindings.begin(); i != bindings.end(); ++i){ (*i)->deliver(msg); } } FanOutExchange::~FanOutExchange() {} const std::string FanOutExchange::typeName("fanout");