View Javadoc

1   //
2   //  ========================================================================
3   //  Copyright (c) 1995-2015 Mort Bay Consulting Pty. Ltd.
4   //  ------------------------------------------------------------------------
5   //  All rights reserved. This program and the accompanying materials
6   //  are made available under the terms of the Eclipse Public License v1.0
7   //  and Apache License v2.0 which accompanies this distribution.
8   //
9   //      The Eclipse Public License is available at
10  //      http://www.eclipse.org/legal/epl-v10.html
11  //
12  //      The Apache License v2.0 is available at
13  //      http://www.opensource.org/licenses/apache2.0.php
14  //
15  //  You may elect to redistribute this code under either of these licenses.
16  //  ========================================================================
17  //
18  
19  package org.eclipse.jetty.client;
20  
21  import java.util.concurrent.atomic.AtomicReference;
22  
23  import org.eclipse.jetty.client.api.Connection;
24  import org.eclipse.jetty.client.api.Request;
25  import org.eclipse.jetty.util.Promise;
26  
27  public abstract class MultiplexHttpDestination<C extends Connection> extends HttpDestination implements Promise<Connection>
28  {
29      private final AtomicReference<ConnectState> connect = new AtomicReference<>(ConnectState.DISCONNECTED);
30      private C connection;
31  
32      protected MultiplexHttpDestination(HttpClient client, Origin origin)
33      {
34          super(client, origin);
35      }
36  
37      @Override
38      protected void send()
39      {
40          while (true)
41          {
42              ConnectState current = connect.get();
43              switch (current)
44              {
45                  case DISCONNECTED:
46                  {
47                      if (!connect.compareAndSet(current, ConnectState.CONNECTING))
48                          break;
49                      newConnection(this);
50                      return;
51                  }
52                  case CONNECTING:
53                  {
54                      // Waiting to connect, just return
55                      return;
56                  }
57                  case CONNECTED:
58                  {
59                      if (process(connection, false))
60                          break;
61                      return;
62                  }
63                  default:
64                  {
65                      abort(new IllegalStateException("Invalid connection state " + current));
66                      return;
67                  }
68              }
69          }
70      }
71  
72      @Override
73      @SuppressWarnings("unchecked")
74      public void succeeded(Connection result)
75      {
76          C connection = this.connection = (C)result;
77          if (connect.compareAndSet(ConnectState.CONNECTING, ConnectState.CONNECTED))
78          {
79              process(connection, true);
80          }
81          else
82          {
83              connection.close();
84              failed(new IllegalStateException());
85          }
86      }
87  
88      @Override
89      public void failed(Throwable x)
90      {
91          connect.set(ConnectState.DISCONNECTED);
92      }
93  
94      protected boolean process(final C connection, boolean dispatch)
95      {
96          HttpClient client = getHttpClient();
97          final HttpExchange exchange = getHttpExchanges().poll();
98          if (LOG.isDebugEnabled())
99              LOG.debug("Processing {} on {}", exchange, connection);
100         if (exchange == null)
101             return false;
102 
103         final Request request = exchange.getRequest();
104         Throwable cause = request.getAbortCause();
105         if (cause != null)
106         {
107             if (LOG.isDebugEnabled())
108                 LOG.debug("Aborted before processing {}: {}", exchange, cause);
109             // It may happen that the request is aborted before the exchange
110             // is created. Aborting the exchange a second time will result in
111             // a no-operation, so we just abort here to cover that edge case.
112             exchange.abort(cause);
113         }
114         else
115         {
116             if (dispatch)
117             {
118                 client.getExecutor().execute(new Runnable()
119                 {
120                     @Override
121                     public void run()
122                     {
123                         send(connection, exchange);
124                     }
125                 });
126             }
127             else
128             {
129                 send(connection, exchange);
130             }
131         }
132         return true;
133     }
134 
135     @Override
136     public void close()
137     {
138         super.close();
139         C connection = this.connection;
140         if (connection != null)
141             connection.close();
142     }
143 
144     @Override
145     public void close(Connection connection)
146     {
147         super.close(connection);
148         while (true)
149         {
150             ConnectState current = connect.get();
151             if (connect.compareAndSet(current, ConnectState.DISCONNECTED))
152             {
153                 if (getHttpClient().isRemoveIdleDestinations())
154                     getHttpClient().removeDestination(this);
155                 break;
156             }
157         }
158     }
159 
160     protected abstract void send(C connection, HttpExchange exchange);
161 
162     private enum ConnectState
163     {
164         DISCONNECTED, CONNECTING, CONNECTED
165     }
166 }