1 /*

     2  * Copyright (c) 2015, 2018, Oracle and/or its affiliates. All rights reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.  Oracle designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Oracle in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    22  * or visit www.oracle.com if you need additional information or have any

    23  * questions.

    24  */

    25 

    26 package jdk.internal.net.http;

    27 

    28 import java.io.IOException;

    29 import java.lang.System.Logger.Level;

    30 import java.net.InetSocketAddress;

    31 import java.net.StandardSocketOptions;

    32 import java.nio.ByteBuffer;

    33 import java.nio.channels.SelectableChannel;

    34 import java.nio.channels.SelectionKey;

    35 import java.nio.channels.SocketChannel;

    36 import java.security.AccessController;

    37 import java.security.PrivilegedActionException;

    38 import java.security.PrivilegedExceptionAction;

    39 import java.util.concurrent.CompletableFuture;

    40 import jdk.internal.net.http.common.FlowTube;

    41 import jdk.internal.net.http.common.Log;

    42 import jdk.internal.net.http.common.MinimalFuture;

    43 import jdk.internal.net.http.common.Utils;

    44 

    45 /**

    46  * Plain raw TCP connection direct to destination.

    47  * The connection operates in asynchronous non-blocking mode.

    48  * All reads and writes are done non-blocking.

    49  */

    50 class PlainHttpConnection extends HttpConnection {

    51 

    52     private final Object reading = new Object();

    53     protected final SocketChannel chan;

    54     private final SocketTube tube; // need SocketTube to call signalClosed().

    55     private final PlainHttpPublisher writePublisher = new PlainHttpPublisher(reading);

    56     private volatile boolean connected;

    57     private boolean closed;

    58 

    59     // should be volatile to provide proper synchronization(visibility) action

    60 

    61     final class ConnectEvent extends AsyncEvent {

    62         private final CompletableFuture<Void> cf;

    63 

    64         ConnectEvent(CompletableFuture<Void> cf) {

    65             this.cf = cf;

    66         }

    67 

    68         @Override

    69         public SelectableChannel channel() {

    70             return chan;

    71         }

    72 

    73         @Override

    74         public int interestOps() {

    75             return SelectionKey.OP_CONNECT;

    76         }

    77 

    78         @Override

    79         public void handle() {

    80             try {

    81                 assert !connected : "Already connected";

    82                 assert !chan.isBlocking() : "Unexpected blocking channel";

    83                 if (debug.on())

    84                     debug.log("ConnectEvent: finishing connect");

    85                 boolean finished = chan.finishConnect();

    86                 assert finished : "Expected channel to be connected";

    87                 if (debug.on())

    88                     debug.log("ConnectEvent: connect finished: %s Local addr: %s",

    89                               finished, chan.getLocalAddress());

    90                 connected = true;

    91                 // complete async since the event runs on the SelectorManager thread

    92                 cf.completeAsync(() -> null, client().theExecutor());

    93             } catch (Throwable e) {

    94                 client().theExecutor().execute( () -> cf.completeExceptionally(e));

    95             }

    96         }

    97 

    98         @Override

    99         public void abort(IOException ioe) {

   100             close();

   101             client().theExecutor().execute( () -> cf.completeExceptionally(ioe));

   102         }

   103     }

   104 

   105     @Override

   106     public CompletableFuture<Void> connectAsync() {

   107         CompletableFuture<Void> cf = new MinimalFuture<>();

   108         try {

   109             assert !connected : "Already connected";

   110             assert !chan.isBlocking() : "Unexpected blocking channel";

   111             boolean finished = false;

   112             PrivilegedExceptionAction<Boolean> pa =

   113                     () -> chan.connect(Utils.resolveAddress(address));

   114             try {

   115                  finished = AccessController.doPrivileged(pa);

   116             } catch (PrivilegedActionException e) {

   117                 cf.completeExceptionally(e.getCause());

   118             }

   119             if (finished) {

   120                 if (debug.on()) debug.log("connect finished without blocking");

   121                 connected = true;

   122                 cf.complete(null);

   123             } else {

   124                 if (debug.on()) debug.log("registering connect event");

   125                 client().registerEvent(new ConnectEvent(cf));

   126             }

   127         } catch (Throwable throwable) {

   128             cf.completeExceptionally(throwable);

   129         }

   130         return cf;

   131     }

   132 

   133     @Override

   134     SocketChannel channel() {

   135         return chan;

   136     }

   137 

   138     @Override

   139     final FlowTube getConnectionFlow() {

   140         return tube;

   141     }

   142 

   143     PlainHttpConnection(InetSocketAddress addr, HttpClientImpl client) {

   144         super(addr, client);

   145         try {

   146             this.chan = SocketChannel.open();

   147             chan.configureBlocking(false);

   148             int bufsize = client.getReceiveBufferSize();

   149             if (!trySetReceiveBufferSize(bufsize)) {

   150                 trySetReceiveBufferSize(256*1024);

   151             }

   152             chan.setOption(StandardSocketOptions.TCP_NODELAY, true);

   153             // wrap the connected channel in a Tube for async reading and writing

   154             tube = new SocketTube(client(), chan, Utils::getBuffer);

   155         } catch (IOException e) {

   156             throw new InternalError(e);

   157         }

   158     }

   159 

   160     private boolean trySetReceiveBufferSize(int bufsize) {

   161         try {

   162             chan.setOption(StandardSocketOptions.SO_RCVBUF, bufsize);

   163             if (debug.on())

   164                 debug.log("Receive buffer size is %s",

   165                           chan.getOption(StandardSocketOptions.SO_RCVBUF));

   166             return true;

   167         } catch(IOException x) {

   168             if (debug.on())

   169                 debug.log("Failed to set receive buffer size to %d on %s",

   170                           bufsize, chan);

   171         }

   172         return false;

   173     }

   174 

   175     @Override

   176     HttpPublisher publisher() { return writePublisher; }

   177 

   178 

   179     @Override

   180     public String toString() {

   181         return "PlainHttpConnection: " + super.toString();

   182     }

   183 

   184     /**

   185      * Closes this connection

   186      */

   187     @Override

   188     public void close() {

   189         synchronized (this) {

   190             if (closed) {

   191                 return;

   192             }

   193             closed = true;

   194         }

   195         try {

   196             Log.logTrace("Closing: " + toString());

   197             if (debug.on())

   198                 debug.log("Closing channel: " + client().debugInterestOps(chan));

   199             chan.close();

   200             tube.signalClosed();

   201         } catch (IOException e) {

   202             Log.logTrace("Closing resulted in " + e);

   203         }

   204     }

   205 

   206 

   207     @Override

   208     ConnectionPool.CacheKey cacheKey() {

   209         return new ConnectionPool.CacheKey(address, null);

   210     }

   211 

   212     @Override

   213     synchronized boolean connected() {

   214         return connected;

   215     }

   216 

   217 

   218     @Override

   219     boolean isSecure() {

   220         return false;

   221     }

   222 

   223     @Override

   224     boolean isProxied() {

   225         return false;

   226     }

   227 

   228 }