KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > org > snmp4j > util > MultiThreadedMessageDispatcher


1 /*_############################################################################
2   _##
3   _## SNMP4J - MultiThreadedMessageDispatcher.java
4   _##
5   _## Copyright 2003-2007 Frank Fock and Jochen Katz (SNMP4J.org)
6   _##
7   _## Licensed under the Apache License, Version 2.0 (the "License");
8   _## you may not use this file except in compliance with the License.
9   _## You may obtain a copy of the License at
10   _##
11   _## http://www.apache.org/licenses/LICENSE-2.0
12   _##
13   _## Unless required by applicable law or agreed to in writing, software
14   _## distributed under the License is distributed on an "AS IS" BASIS,
15   _## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   _## See the License for the specific language governing permissions and
17   _## limitations under the License.
18   _##
19   _##########################################################################*/
20
21
22
23 package org.snmp4j.util;
24
25 import org.snmp4j.MessageDispatcher;
26 import org.snmp4j.CommandResponder;
27 import org.snmp4j.mp.MessageProcessingModel;
28 import org.snmp4j.TransportMapping;
29 import java.util.Collection JavaDoc;
30 import org.snmp4j.smi.Address;
31 import org.snmp4j.asn1.BERInputStream;
32 import org.snmp4j.mp.PduHandle;
33 import org.snmp4j.PDU;
34 import org.snmp4j.mp.StateReference;
35 import org.snmp4j.mp.StatusInformation;
36 import org.snmp4j.MessageException;
37 import java.nio.ByteBuffer JavaDoc;
38 import org.snmp4j.mp.PduHandleCallback;
39
40 /**
41  * The <code>MultiThreadedMessageDispatcher</code> class is a decorator
42  * for any <code>MessageDispatcher</code> instances that processes incoming
43  * message with a supplied <code>ThreadPool</code>. The processing is thus
44  * parallelized on up to the size of the supplied thread pool threads.
45  *
46  * @author Frank Fock
47  * @version 1.8
48  * @since 1.0.2
49  */
50 public class MultiThreadedMessageDispatcher implements MessageDispatcher {
51
52   private MessageDispatcher dispatcher;
53   private ThreadPool threadPool;
54
55   /**
56    * Creates a multi thread message dispatcher using the provided
57    * <code>ThreadPool</code> to concurrently process incoming messages
58    * that are forwarded to the supplied decorated
59    * <code>MessageDispatcher</code>.
60    *
61    * @param threadPool
62    * a <code>ThreadPool</code> instance (that can be shared). <em>The thread
63    * pool has to be stopped externally.</em>
64    * @param decoratedDispatcher
65    * the decorated <code>MessageDispatcher</code> that must be
66    * multi-threading safe.
67    */
68   public MultiThreadedMessageDispatcher(ThreadPool threadPool,
69                                         MessageDispatcher decoratedDispatcher) {
70     this.threadPool = threadPool;
71     this.dispatcher = decoratedDispatcher;
72   }
73
74   public int getNextRequestID() {
75     return dispatcher.getNextRequestID();
76   }
77
78   public void addMessageProcessingModel(MessageProcessingModel model) {
79     dispatcher.addMessageProcessingModel(model);
80   }
81
82   public void removeMessageProcessingModel(MessageProcessingModel model) {
83     dispatcher.removeMessageProcessingModel(model);
84   }
85
86   public MessageProcessingModel getMessageProcessingModel(int messageProcessingModel) {
87     return dispatcher.getMessageProcessingModel(messageProcessingModel);
88   }
89
90   public void addTransportMapping(TransportMapping transport) {
91     dispatcher.addTransportMapping(transport);
92   }
93
94   public TransportMapping removeTransportMapping(TransportMapping transport) {
95     return dispatcher.removeTransportMapping(transport);
96   }
97
98   public Collection JavaDoc getTransportMappings() {
99     return dispatcher.getTransportMappings();
100   }
101
102   public void addCommandResponder(CommandResponder listener) {
103     dispatcher.addCommandResponder(listener);
104   }
105
106   public void removeCommandResponder(CommandResponder listener) {
107     dispatcher.removeCommandResponder(listener);
108   }
109
110   public PduHandle sendPdu(Address transportAddress,
111                            int messageProcessingModel,
112                            int securityModel,
113                            byte[] securityName,
114                            int securityLevel,
115                            PDU pdu,
116                            boolean expectResponse) throws MessageException {
117     return dispatcher.sendPdu(transportAddress, messageProcessingModel,
118                               securityModel, securityName, securityLevel,
119                               pdu, expectResponse);
120   }
121
122   public PduHandle sendPdu(TransportMapping transportMapping,
123                            Address transportAddress,
124                            int messageProcessingModel,
125                            int securityModel,
126                            byte[] securityName,
127                            int securityLevel,
128                            PDU pdu,
129                            boolean expectResponse) throws MessageException {
130     return dispatcher.sendPdu(transportMapping, transportAddress,
131                               messageProcessingModel,
132                               securityModel, securityName,
133                               securityLevel, pdu, expectResponse);
134   }
135
136   public PduHandle sendPdu(TransportMapping transportMapping,
137                            Address transportAddress,
138                            int messageProcessingModel,
139                            int securityModel, byte[] securityName,
140                            int securityLevel, PDU pdu, boolean expectResponse,
141                            PduHandleCallback callback) throws MessageException {
142     return dispatcher.sendPdu(transportMapping, transportAddress,
143                               messageProcessingModel,
144                               securityModel, securityName,
145                               securityLevel, pdu, expectResponse, callback);
146   }
147
148   public int returnResponsePdu(int messageProcessingModel,
149                                int securityModel,
150                                byte[] securityName,
151                                int securityLevel,
152                                PDU pdu,
153                                int maxSizeResponseScopedPDU,
154                                StateReference stateReference,
155                                StatusInformation statusInformation)
156       throws MessageException
157   {
158     return dispatcher.returnResponsePdu(messageProcessingModel,
159                                         securityModel, securityName,
160                                         securityLevel, pdu,
161                                         maxSizeResponseScopedPDU,
162                                         stateReference,
163                                         statusInformation);
164   }
165
166   public void processMessage(TransportMapping sourceTransport,
167                              Address incomingAddress,
168                              BERInputStream wholeMessage) {
169     // OK, here wo do all that what this class is all about!
170 MessageTask task = new MessageTask(sourceTransport,
171                                        incomingAddress,
172                                        wholeMessage);
173     threadPool.execute(task);
174   }
175
176   public void processMessage(TransportMapping sourceTransport,
177                              Address incomingAddress, ByteBuffer JavaDoc wholeMessage) {
178     processMessage(sourceTransport, incomingAddress,
179                    new BERInputStream(wholeMessage));
180   }
181
182   public void releaseStateReference(int messageProcessingModel,
183                                     PduHandle pduHandle) {
184     dispatcher.releaseStateReference(messageProcessingModel, pduHandle);
185   }
186
187   public TransportMapping getTransport(Address destAddress) {
188     return dispatcher.getTransport(destAddress);
189   }
190
191   class MessageTask implements Runnable JavaDoc {
192     private TransportMapping sourceTransport;
193     private Address incomingAddress;
194     private BERInputStream wholeMessage;
195
196     public MessageTask(TransportMapping sourceTransport,
197                        Address incomingAddress,
198                        BERInputStream wholeMessage) {
199       this.sourceTransport = sourceTransport;
200       this.incomingAddress = incomingAddress;
201       this.wholeMessage = wholeMessage;
202     }
203
204     public void run() {
205       dispatcher.processMessage(sourceTransport, incomingAddress, wholeMessage);
206     }
207
208   }
209 }
210
Popular Tags