View Javadoc
1   /*
2    * Copyright 2012 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License,
5    * version 2.0 (the "License"); you may not use this file except in compliance
6    * with the License. You may obtain a copy of the License at:
7    *
8    *   https://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13   * License for the specific language governing permissions and limitations
14   * under the License.
15   */
16  
17  package io.netty.bootstrap;
18  
19  import io.netty.channel.Channel;
20  import io.netty.channel.ChannelFuture;
21  import io.netty.channel.ChannelFutureListener;
22  import io.netty.channel.ChannelHandler;
23  import io.netty.channel.ChannelOption;
24  import io.netty.channel.ChannelPromise;
25  import io.netty.channel.DefaultChannelPromise;
26  import io.netty.channel.EventLoop;
27  import io.netty.channel.EventLoopGroup;
28  import io.netty.channel.ReflectiveChannelFactory;
29  import io.netty.util.AttributeKey;
30  import io.netty.util.concurrent.EventExecutor;
31  import io.netty.util.concurrent.GlobalEventExecutor;
32  import io.netty.util.internal.ObjectUtil;
33  import io.netty.util.internal.SocketUtils;
34  import io.netty.util.internal.StringUtil;
35  import io.netty.util.internal.logging.InternalLogger;
36  
37  import java.net.InetAddress;
38  import java.net.InetSocketAddress;
39  import java.net.SocketAddress;
40  import java.util.Collection;
41  import java.util.Collections;
42  import java.util.HashMap;
43  import java.util.LinkedHashMap;
44  import java.util.Map;
45  import java.util.concurrent.ConcurrentHashMap;
46  
47  /**
48   * {@link AbstractBootstrap} is a helper class that makes it easy to bootstrap a {@link Channel}. It support
49   * method-chaining to provide an easy way to configure the {@link AbstractBootstrap}.
50   *
51   * <p>When not used in a {@link ServerBootstrap} context, the {@link #bind()} methods are useful for connectionless
52   * transports such as datagram (UDP).</p>
53   */
54  public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C extends Channel> implements Cloneable {
55      @SuppressWarnings("unchecked")
56      private static final Map.Entry<ChannelOption<?>, Object>[] EMPTY_OPTION_ARRAY = new Map.Entry[0];
57      @SuppressWarnings("unchecked")
58      private static final Map.Entry<AttributeKey<?>, Object>[] EMPTY_ATTRIBUTE_ARRAY = new Map.Entry[0];
59  
60      volatile EventLoopGroup group;
61      @SuppressWarnings("deprecation")
62      private volatile ChannelFactory<? extends C> channelFactory;
63      private volatile SocketAddress localAddress;
64  
65      // The order in which ChannelOptions are applied is important they may depend on each other for validation
66      // purposes.
67      private final Map<ChannelOption<?>, Object> options = new LinkedHashMap<ChannelOption<?>, Object>();
68      private final Map<AttributeKey<?>, Object> attrs = new ConcurrentHashMap<AttributeKey<?>, Object>();
69      private volatile ChannelHandler handler;
70      private volatile ClassLoader extensionsClassLoader;
71  
72      AbstractBootstrap() {
73          // Disallow extending from a different package.
74      }
75  
76      AbstractBootstrap(AbstractBootstrap<B, C> bootstrap) {
77          group = bootstrap.group;
78          channelFactory = bootstrap.channelFactory;
79          handler = bootstrap.handler;
80          localAddress = bootstrap.localAddress;
81          synchronized (bootstrap.options) {
82              options.putAll(bootstrap.options);
83          }
84          attrs.putAll(bootstrap.attrs);
85          extensionsClassLoader = bootstrap.extensionsClassLoader;
86      }
87  
88      /**
89       * The {@link EventLoopGroup} which is used to handle all the events for the to-be-created
90       * {@link Channel}
91       */
92      public B group(EventLoopGroup group) {
93          ObjectUtil.checkNotNull(group, "group");
94          if (this.group != null) {
95              throw new IllegalStateException("group set already");
96          }
97          this.group = group;
98          return self();
99      }
100 
101     @SuppressWarnings("unchecked")
102     private B self() {
103         return (B) this;
104     }
105 
106     /**
107      * The {@link Class} which is used to create {@link Channel} instances from.
108      * You either use this or {@link #channelFactory(io.netty.channel.ChannelFactory)} if your
109      * {@link Channel} implementation has no no-args constructor.
110      */
111     public B channel(Class<? extends C> channelClass) {
112         return channelFactory(new ReflectiveChannelFactory<C>(
113                 ObjectUtil.checkNotNull(channelClass, "channelClass")
114         ));
115     }
116 
117     /**
118      * @deprecated Use {@link #channelFactory(io.netty.channel.ChannelFactory)} instead.
119      */
120     @Deprecated
121     public B channelFactory(ChannelFactory<? extends C> channelFactory) {
122         ObjectUtil.checkNotNull(channelFactory, "channelFactory");
123         if (this.channelFactory != null) {
124             throw new IllegalStateException("channelFactory set already");
125         }
126 
127         this.channelFactory = channelFactory;
128         return self();
129     }
130 
131     /**
132      * {@link io.netty.channel.ChannelFactory} which is used to create {@link Channel} instances from
133      * when calling {@link #bind()}. This method is usually only used if {@link #channel(Class)}
134      * is not working for you because of some more complex needs. If your {@link Channel} implementation
135      * has a no-args constructor, its highly recommend to just use {@link #channel(Class)} to
136      * simplify your code.
137      */
138     @SuppressWarnings({ "unchecked", "deprecation" })
139     public B channelFactory(io.netty.channel.ChannelFactory<? extends C> channelFactory) {
140         return channelFactory((ChannelFactory<C>) channelFactory);
141     }
142 
143     /**
144      * The {@link SocketAddress} which is used to bind the local "end" to.
145      */
146     public B localAddress(SocketAddress localAddress) {
147         this.localAddress = localAddress;
148         return self();
149     }
150 
151     /**
152      * @see #localAddress(SocketAddress)
153      */
154     public B localAddress(int inetPort) {
155         return localAddress(new InetSocketAddress(inetPort));
156     }
157 
158     /**
159      * @see #localAddress(SocketAddress)
160      */
161     public B localAddress(String inetHost, int inetPort) {
162         return localAddress(SocketUtils.socketAddress(inetHost, inetPort));
163     }
164 
165     /**
166      * @see #localAddress(SocketAddress)
167      */
168     public B localAddress(InetAddress inetHost, int inetPort) {
169         return localAddress(new InetSocketAddress(inetHost, inetPort));
170     }
171 
172     /**
173      * Allow to specify a {@link ChannelOption} which is used for the {@link Channel} instances once they got
174      * created. Use a value of {@code null} to remove a previous set {@link ChannelOption}.
175      */
176     public <T> B option(ChannelOption<T> option, T value) {
177         ObjectUtil.checkNotNull(option, "option");
178         synchronized (options) {
179             if (value == null) {
180                 options.remove(option);
181             } else {
182                 options.put(option, value);
183             }
184         }
185         return self();
186     }
187 
188     /**
189      * Allow to specify an initial attribute of the newly created {@link Channel}.  If the {@code value} is
190      * {@code null}, the attribute of the specified {@code key} is removed.
191      */
192     public <T> B attr(AttributeKey<T> key, T value) {
193         ObjectUtil.checkNotNull(key, "key");
194         if (value == null) {
195             attrs.remove(key);
196         } else {
197             attrs.put(key, value);
198         }
199         return self();
200     }
201 
202     /**
203      * Load {@link ChannelInitializerExtension}s using the given class loader.
204      * <p>
205      * By default, the extensions will be loaded by the same class loader that loaded this bootstrap class.
206      *
207      * @param classLoader The class loader to use for loading {@link ChannelInitializerExtension}s.
208      * @return This bootstrap.
209      */
210     public B extensionsClassLoader(ClassLoader classLoader) {
211         extensionsClassLoader = classLoader;
212         return self();
213     }
214 
215     /**
216      * Validate all the parameters. Sub-classes may override this, but should
217      * call the super method in that case.
218      */
219     public B validate() {
220         if (group == null) {
221             throw new IllegalStateException("group not set");
222         }
223         if (channelFactory == null) {
224             throw new IllegalStateException("channel or channelFactory not set");
225         }
226         return self();
227     }
228 
229     /**
230      * Returns a deep clone of this bootstrap which has the identical configuration.  This method is useful when making
231      * multiple {@link Channel}s with similar settings.  Please note that this method does not clone the
232      * {@link EventLoopGroup} deeply but shallowly, making the group a shared resource.
233      */
234     @Override
235     @SuppressWarnings("CloneDoesntDeclareCloneNotSupportedException")
236     public abstract B clone();
237 
238     /**
239      * Create a new {@link Channel} and register it with an {@link EventLoop}.
240      */
241     public ChannelFuture register() {
242         validate();
243         return initAndRegister();
244     }
245 
246     /**
247      * Create a new {@link Channel} and bind it.
248      */
249     public ChannelFuture bind() {
250         validate();
251         SocketAddress localAddress = this.localAddress;
252         if (localAddress == null) {
253             throw new IllegalStateException("localAddress not set");
254         }
255         return doBind(localAddress);
256     }
257 
258     /**
259      * Create a new {@link Channel} and bind it.
260      */
261     public ChannelFuture bind(int inetPort) {
262         return bind(new InetSocketAddress(inetPort));
263     }
264 
265     /**
266      * Create a new {@link Channel} and bind it.
267      */
268     public ChannelFuture bind(String inetHost, int inetPort) {
269         return bind(SocketUtils.socketAddress(inetHost, inetPort));
270     }
271 
272     /**
273      * Create a new {@link Channel} and bind it.
274      */
275     public ChannelFuture bind(InetAddress inetHost, int inetPort) {
276         return bind(new InetSocketAddress(inetHost, inetPort));
277     }
278 
279     /**
280      * Create a new {@link Channel} and bind it.
281      */
282     public ChannelFuture bind(SocketAddress localAddress) {
283         validate();
284         return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress"));
285     }
286 
287     private ChannelFuture doBind(final SocketAddress localAddress) {
288         final ChannelFuture regFuture = initAndRegister();
289         final Channel channel = regFuture.channel();
290         if (regFuture.cause() != null) {
291             return regFuture;
292         }
293 
294         if (regFuture.isDone()) {
295             // At this point we know that the registration was complete and successful.
296             ChannelPromise promise = channel.newPromise();
297             doBind0(regFuture, channel, localAddress, promise);
298             return promise;
299         } else {
300             // Registration future is almost always fulfilled already, but just in case it's not.
301             final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
302             regFuture.addListener(new ChannelFutureListener() {
303                 @Override
304                 public void operationComplete(ChannelFuture future) throws Exception {
305                     Throwable cause = future.cause();
306                     if (cause != null) {
307                         // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
308                         // IllegalStateException once we try to access the EventLoop of the Channel.
309                         promise.setFailure(cause);
310                     } else {
311                         // Registration was successful, so set the correct executor to use.
312                         // See https://github.com/netty/netty/issues/2586
313                         promise.registered();
314 
315                         doBind0(regFuture, channel, localAddress, promise);
316                     }
317                 }
318             });
319             return promise;
320         }
321     }
322 
323     final ChannelFuture initAndRegister() {
324         Channel channel = null;
325         try {
326             channel = channelFactory.newChannel();
327             init(channel);
328         } catch (Throwable t) {
329             if (channel != null) {
330                 // channel can be null if newChannel crashed (eg SocketException("too many open files"))
331                 channel.unsafe().closeForcibly();
332                 // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
333                 return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
334             }
335             // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
336             return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
337         }
338 
339         ChannelFuture regFuture = config().group().register(channel);
340         if (regFuture.cause() != null) {
341             if (channel.isRegistered()) {
342                 channel.close();
343             } else {
344                 channel.unsafe().closeForcibly();
345             }
346         }
347 
348         // If we are here and the promise is not failed, it's one of the following cases:
349         // 1) If we attempted registration from the event loop, the registration has been completed at this point.
350         //    i.e. It's safe to attempt bind() or connect() now because the channel has been registered.
351         // 2) If we attempted registration from the other thread, the registration request has been successfully
352         //    added to the event loop's task queue for later execution.
353         //    i.e. It's safe to attempt bind() or connect() now:
354         //         because bind() or connect() will be executed *after* the scheduled registration task is executed
355         //         because register(), bind(), and connect() are all bound to the same thread.
356 
357         return regFuture;
358     }
359 
360     abstract void init(Channel channel) throws Exception;
361 
362     Collection<ChannelInitializerExtension> getInitializerExtensions() {
363         ClassLoader loader = extensionsClassLoader;
364         if (loader == null) {
365             loader = getClass().getClassLoader();
366         }
367         return ChannelInitializerExtensions.getExtensions().extensions(loader);
368     }
369 
370     private static void doBind0(
371             final ChannelFuture regFuture, final Channel channel,
372             final SocketAddress localAddress, final ChannelPromise promise) {
373 
374         // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
375         // the pipeline in its channelRegistered() implementation.
376         channel.eventLoop().execute(new Runnable() {
377             @Override
378             public void run() {
379                 if (regFuture.isSuccess()) {
380                     channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
381                 } else {
382                     promise.setFailure(regFuture.cause());
383                 }
384             }
385         });
386     }
387 
388     /**
389      * the {@link ChannelHandler} to use for serving the requests.
390      */
391     public B handler(ChannelHandler handler) {
392         this.handler = ObjectUtil.checkNotNull(handler, "handler");
393         return self();
394     }
395 
396     /**
397      * Returns the configured {@link EventLoopGroup} or {@code null} if non is configured yet.
398      *
399      * @deprecated Use {@link #config()} instead.
400      */
401     @Deprecated
402     public final EventLoopGroup group() {
403         return group;
404     }
405 
406     /**
407      * Returns the {@link AbstractBootstrapConfig} object that can be used to obtain the current config
408      * of the bootstrap.
409      */
410     public abstract AbstractBootstrapConfig<B, C> config();
411 
412     final Map.Entry<ChannelOption<?>, Object>[] newOptionsArray() {
413         return newOptionsArray(options);
414     }
415 
416     static Map.Entry<ChannelOption<?>, Object>[] newOptionsArray(Map<ChannelOption<?>, Object> options) {
417         synchronized (options) {
418             return new LinkedHashMap<ChannelOption<?>, Object>(options).entrySet().toArray(EMPTY_OPTION_ARRAY);
419         }
420     }
421 
422     final Map.Entry<AttributeKey<?>, Object>[] newAttributesArray() {
423         return newAttributesArray(attrs0());
424     }
425 
426     static Map.Entry<AttributeKey<?>, Object>[] newAttributesArray(Map<AttributeKey<?>, Object> attributes) {
427         return attributes.entrySet().toArray(EMPTY_ATTRIBUTE_ARRAY);
428     }
429 
430     final Map<ChannelOption<?>, Object> options0() {
431         return options;
432     }
433 
434     final Map<AttributeKey<?>, Object> attrs0() {
435         return attrs;
436     }
437 
438     final SocketAddress localAddress() {
439         return localAddress;
440     }
441 
442     @SuppressWarnings("deprecation")
443     final ChannelFactory<? extends C> channelFactory() {
444         return channelFactory;
445     }
446 
447     final ChannelHandler handler() {
448         return handler;
449     }
450 
451     final Map<ChannelOption<?>, Object> options() {
452         synchronized (options) {
453             return copiedMap(options);
454         }
455     }
456 
457     final Map<AttributeKey<?>, Object> attrs() {
458         return copiedMap(attrs);
459     }
460 
461     static <K, V> Map<K, V> copiedMap(Map<K, V> map) {
462         if (map.isEmpty()) {
463             return Collections.emptyMap();
464         }
465         return Collections.unmodifiableMap(new HashMap<K, V>(map));
466     }
467 
468     static void setAttributes(Channel channel, Map.Entry<AttributeKey<?>, Object>[] attrs) {
469         for (Map.Entry<AttributeKey<?>, Object> e: attrs) {
470             @SuppressWarnings("unchecked")
471             AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
472             channel.attr(key).set(e.getValue());
473         }
474     }
475 
476     static void setChannelOptions(
477             Channel channel, Map.Entry<ChannelOption<?>, Object>[] options, InternalLogger logger) {
478         for (Map.Entry<ChannelOption<?>, Object> e: options) {
479             setChannelOption(channel, e.getKey(), e.getValue(), logger);
480         }
481     }
482 
483     @SuppressWarnings("unchecked")
484     private static void setChannelOption(
485             Channel channel, ChannelOption<?> option, Object value, InternalLogger logger) {
486         try {
487             if (!channel.config().setOption((ChannelOption<Object>) option, value)) {
488                 logger.warn("Unknown channel option '{}' for channel '{}'", option, channel);
489             }
490         } catch (Throwable t) {
491             logger.warn(
492                     "Failed to set channel option '{}' with value '{}' for channel '{}'", option, value, channel, t);
493         }
494     }
495 
496     @Override
497     public String toString() {
498         StringBuilder buf = new StringBuilder()
499             .append(StringUtil.simpleClassName(this))
500             .append('(').append(config()).append(')');
501         return buf.toString();
502     }
503 
504     static final class PendingRegistrationPromise extends DefaultChannelPromise {
505 
506         // Is set to the correct EventExecutor once the registration was successful. Otherwise it will
507         // stay null and so the GlobalEventExecutor.INSTANCE will be used for notifications.
508         private volatile boolean registered;
509 
510         PendingRegistrationPromise(Channel channel) {
511             super(channel);
512         }
513 
514         void registered() {
515             registered = true;
516         }
517 
518         @Override
519         protected EventExecutor executor() {
520             if (registered) {
521                 // If the registration was a success executor is set.
522                 //
523                 // See https://github.com/netty/netty/issues/2586
524                 return super.executor();
525             }
526             // The registration failed so we can only use the GlobalEventExecutor as last resort to notify.
527             return GlobalEventExecutor.INSTANCE;
528         }
529     }
530 }