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