001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.activemq.broker;
018
019import java.net.URI;
020import java.util.Map;
021import java.util.Set;
022import java.util.concurrent.ThreadPoolExecutor;
023
024import org.apache.activemq.Service;
025import org.apache.activemq.broker.region.Destination;
026import org.apache.activemq.broker.region.MessageReference;
027import org.apache.activemq.broker.region.Region;
028import org.apache.activemq.broker.region.Subscription;
029import org.apache.activemq.command.ActiveMQDestination;
030import org.apache.activemq.command.BrokerId;
031import org.apache.activemq.command.BrokerInfo;
032import org.apache.activemq.command.ConnectionInfo;
033import org.apache.activemq.command.DestinationInfo;
034import org.apache.activemq.command.MessageDispatch;
035import org.apache.activemq.command.ProducerInfo;
036import org.apache.activemq.command.SessionInfo;
037import org.apache.activemq.command.TransactionId;
038import org.apache.activemq.store.PListStore;
039import org.apache.activemq.thread.Scheduler;
040import org.apache.activemq.usage.Usage;
041
042/**
043 * The Message Broker which routes messages, maintains subscriptions and
044 * connections, acknowledges messages and handles transactions.
045 *
046 *
047 */
048public interface Broker extends Region, Service {
049
050    /**
051     * Get a Broker from the Broker Stack that is a particular class
052     *
053     * @param type
054     * @return
055     */
056    Broker getAdaptor(Class type);
057
058    /**
059     * Get the id of the broker
060     */
061    BrokerId getBrokerId();
062
063    /**
064     * Get the name of the broker
065     */
066    String getBrokerName();
067
068    /**
069     * A remote Broker connects
070     */
071    void addBroker(Connection connection, BrokerInfo info);
072
073    /**
074     * Remove a BrokerInfo
075     *
076     * @param connection
077     * @param info
078     */
079    void removeBroker(Connection connection, BrokerInfo info);
080
081    /**
082     * A client is establishing a connection with the broker.
083     *
084     * @throws Exception TODO
085     */
086    void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception;
087
088    /**
089     * A client is disconnecting from the broker.
090     *
091     * @param context the environment the operation is being executed under.
092     * @param info
093     * @param error null if the client requested the disconnect or the error
094     *                that caused the client to disconnect.
095     * @throws Exception TODO
096     */
097    void removeConnection(ConnectionContext context, ConnectionInfo info, Throwable error) throws Exception;
098
099    /**
100     * Adds a session.
101     *
102     * @param context
103     * @param info
104     * @throws Exception TODO
105     */
106    void addSession(ConnectionContext context, SessionInfo info) throws Exception;
107
108    /**
109     * Removes a session.
110     *
111     * @param context
112     * @param info
113     * @throws Exception TODO
114     */
115    void removeSession(ConnectionContext context, SessionInfo info) throws Exception;
116
117    /**
118     * Adds a producer.
119     *
120     * @param context the enviorment the operation is being executed under.
121     * @throws Exception TODO
122     */
123    @Override
124    void addProducer(ConnectionContext context, ProducerInfo info) throws Exception;
125
126    /**
127     * Removes a producer.
128     *
129     * @param context the enviorment the operation is being executed under.
130     * @throws Exception TODO
131     */
132    @Override
133    void removeProducer(ConnectionContext context, ProducerInfo info) throws Exception;
134
135    /**
136     * @return all clients added to the Broker.
137     * @throws Exception TODO
138     */
139    Connection[] getClients() throws Exception;
140
141    /**
142     * @return all destinations added to the Broker.
143     * @throws Exception TODO
144     */
145    ActiveMQDestination[] getDestinations() throws Exception;
146
147    /**
148     * return a reference destination map of a region based on the destination type
149     * @param destination
150     * @return
151     */
152    public Map<ActiveMQDestination, Destination> getDestinationMap(ActiveMQDestination destination);
153
154    /**
155     * Gets a list of all the prepared xa transactions.
156     *
157     * @param context transaction ids
158     * @return
159     * @throws Exception TODO
160     */
161    TransactionId[] getPreparedTransactions(ConnectionContext context) throws Exception;
162
163    /**
164     * Starts a transaction.
165     *
166     * @param context
167     * @param xid
168     * @throws Exception TODO
169     */
170    void beginTransaction(ConnectionContext context, TransactionId xid) throws Exception;
171
172    /**
173     * Prepares a transaction. Only valid for xa transactions.
174     *
175     * @param context
176     * @param xid
177     * @return id
178     * @throws Exception TODO
179     */
180    int prepareTransaction(ConnectionContext context, TransactionId xid) throws Exception;
181
182    /**
183     * Rollsback a transaction.
184     *
185     * @param context
186     * @param xid
187     * @throws Exception TODO
188     */
189
190    void rollbackTransaction(ConnectionContext context, TransactionId xid) throws Exception;
191
192    /**
193     * Commits a transaction.
194     *
195     * @param context
196     * @param xid
197     * @param onePhase
198     * @throws Exception TODO
199     */
200    void commitTransaction(ConnectionContext context, TransactionId xid, boolean onePhase) throws Exception;
201
202    /**
203     * Forgets a transaction.
204     *
205     * @param context
206     * @param transactionId
207     * @throws Exception
208     */
209    void forgetTransaction(ConnectionContext context, TransactionId transactionId) throws Exception;
210
211    /**
212     * Get the BrokerInfo's of any connected Brokers
213     *
214     * @return array of peer BrokerInfos
215     */
216    BrokerInfo[] getPeerBrokerInfos();
217
218    /**
219     * Notify the Broker that a dispatch is going to happen
220     *
221     * @param messageDispatch
222     */
223    void preProcessDispatch(MessageDispatch messageDispatch);
224
225    /**
226     * Notify the Broker that a dispatch has happened
227     *
228     * @param messageDispatch
229     */
230    void postProcessDispatch(MessageDispatch messageDispatch);
231
232    /**
233     * @return true if the broker has stopped
234     */
235    boolean isStopped();
236
237    /**
238     * @return a Set of all durable destinations
239     */
240    Set<ActiveMQDestination> getDurableDestinations();
241
242    /**
243     * Add and process a DestinationInfo object
244     *
245     * @param context
246     * @param info
247     * @throws Exception
248     */
249    void addDestinationInfo(ConnectionContext context, DestinationInfo info) throws Exception;
250
251    /**
252     * Remove and process a DestinationInfo object
253     *
254     * @param context
255     * @param info
256     * @throws Exception
257     */
258    void removeDestinationInfo(ConnectionContext context, DestinationInfo info) throws Exception;
259
260    /**
261     * @return true if fault tolerant
262     */
263    boolean isFaultTolerantConfiguration();
264
265    /**
266     * @return the connection context used to make administration operations on
267     *         startup or via JMX MBeans
268     */
269    ConnectionContext getAdminConnectionContext();
270
271    /**
272     * Sets the default administration connection context used when configuring
273     * the broker on startup or via JMX
274     *
275     * @param adminConnectionContext
276     */
277    void setAdminConnectionContext(ConnectionContext adminConnectionContext);
278
279    /**
280     * @return the temp data store
281     */
282    PListStore getTempDataStore();
283
284    /**
285     * @return the URI that can be used to connect to the local Broker
286     */
287    URI getVmConnectorURI();
288
289    /**
290     * called when the brokerService starts
291     */
292    void brokerServiceStarted();
293
294    /**
295     * @return the BrokerService
296     */
297    BrokerService getBrokerService();
298
299    /**
300     * Ensure we get the Broker at the top of the Stack
301     *
302     * @return the broker at the top of the Stack
303     */
304    Broker getRoot();
305
306    /**
307     * Determine if a message has expired -allows default behaviour to be
308     * overriden - as the timestamp set by the producer can be out of sync with
309     * the broker
310     *
311     * @param messageReference
312     * @return true if the message is expired
313     */
314    boolean isExpired(MessageReference messageReference);
315
316    /**
317     * A Message has Expired
318     *
319     * @param context
320     * @param messageReference
321     * @param subscription, may be null
322     */
323    void messageExpired(ConnectionContext context, MessageReference messageReference, Subscription subscription);
324
325    /**
326     * A message needs to go the a DLQ
327     *
328     *
329     * @param context
330     * @param messageReference
331     * @param poisonCause reason for dlq submission, may be null
332     * @return true if Message was placed in a DLQ false if discarded.
333     */
334    boolean sendToDeadLetterQueue(ConnectionContext context, MessageReference messageReference, Subscription subscription, Throwable poisonCause);
335
336    /**
337     * @return the broker sequence id
338     */
339    long getBrokerSequenceId();
340
341    /**
342     * called when message is consumed
343     * @param context
344     * @param messageReference
345     */
346    void messageConsumed(ConnectionContext context, MessageReference messageReference);
347
348    /**
349     * Called when message is delivered to the broker
350     * @param context
351     * @param messageReference
352     */
353    void messageDelivered(ConnectionContext context, MessageReference messageReference);
354
355    /**
356     * Called when a message is discarded - e.g. running low on memory
357     * This will happen only if the policy is enabled - e.g. non durable topics
358     * @param context
359     * @param sub
360     * @param messageReference
361     */
362    void messageDiscarded(ConnectionContext context, Subscription sub, MessageReference messageReference);
363
364    /**
365     * Called when there is a slow consumer
366     * @param context
367     * @param destination
368     * @param subs
369     */
370    void slowConsumer(ConnectionContext context,Destination destination, Subscription subs);
371
372    /**
373     * Called to notify a producer is too fast
374     * @param context
375     * @param producerInfo
376     * @param destination
377     */
378    void fastProducer(ConnectionContext context,ProducerInfo producerInfo,ActiveMQDestination destination);
379
380    /**
381     * Called when a Usage reaches a limit
382     * @param context
383     * @param destination
384     * @param usage
385     */
386    void isFull(ConnectionContext context,Destination destination,Usage usage);
387
388    /**
389     *  called when the broker becomes the master in a master/slave
390     *  configuration
391     */
392    void nowMasterBroker();
393
394    Scheduler getScheduler();
395
396    ThreadPoolExecutor getExecutor();
397
398    void networkBridgeStarted(BrokerInfo brokerInfo, boolean createdByDuplex, String remoteIp);
399
400    void networkBridgeStopped(BrokerInfo brokerInfo);
401
402
403}