hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableImpl.java

   1 /**
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements.  See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership.  The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License.  You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  */
  18 package org.apache.hadoop.hbase.client;
  19
  20 import static java.util.stream.Collectors.toList;
  21
  22 import java.io.IOException;
  23 import java.util.List;
  24 import java.util.concurrent.CompletableFuture;
  25 import java.util.concurrent.ExecutorService;
  26 import java.util.concurrent.TimeUnit;
  27 import java.util.function.Function;
  28 import org.apache.hadoop.conf.Configuration;
  29 import org.apache.hadoop.hbase.CompareOperator;
  30 import org.apache.hadoop.hbase.TableName;
  31 import org.apache.hadoop.hbase.filter.Filter;
  32 import org.apache.hadoop.hbase.io.TimeRange;
  33 import org.apache.hadoop.hbase.util.FutureUtils;
  34 import org.apache.yetus.audience.InterfaceAudience;
  35
  36 import org.apache.hbase.thirdparty.com.google.protobuf.RpcChannel;
  37
  38 /**
  39  * Just a wrapper of {@link RawAsyncTableImpl}. The difference is that users need to provide a
  40  * thread pool when constructing this class, and the callback methods registered to the returned
  41  * {@link CompletableFuture} will be executed in this thread pool. So usually it is safe for users
  42  * to do anything they want in the callbacks without breaking the rpc framework.
  43  */
  44 @InterfaceAudience.Private
  45 class AsyncTableImpl implements AsyncTable<ScanResultConsumer> {
  46
  47   private final RawAsyncTableImpl rawTable;
  48
  49   private final ExecutorService pool;
  50
  51   AsyncTableImpl(RawAsyncTableImpl rawTable, ExecutorService pool) {
  52     this.rawTable = rawTable;
  53     this.pool = pool;
  54   }
  55
  56   @Override
  57   public TableName getName() {
  58     return rawTable.getName();
  59   }
  60
  61   @Override
  62   public Configuration getConfiguration() {
  63     return rawTable.getConfiguration();
  64   }
  65
  66   @Override
  67   public CompletableFuture<TableDescriptor> getDescriptor() {
  68     return wrap(rawTable.getDescriptor());
  69   }
  70
  71   @Override
  72   public AsyncTableRegionLocator getRegionLocator() {
  73     return rawTable.getRegionLocator();
  74   }
  75
  76   @Override
  77   public long getRpcTimeout(TimeUnit unit) {
  78     return rawTable.getRpcTimeout(unit);
  79   }
  80
  81   @Override
  82   public long getReadRpcTimeout(TimeUnit unit) {
  83     return rawTable.getReadRpcTimeout(unit);
  84   }
  85
  86   @Override
  87   public long getWriteRpcTimeout(TimeUnit unit) {
  88     return rawTable.getWriteRpcTimeout(unit);
  89   }
  90
  91   @Override
  92   public long getOperationTimeout(TimeUnit unit) {
  93     return rawTable.getOperationTimeout(unit);
  94   }
  95
  96   @Override
  97   public long getScanTimeout(TimeUnit unit) {
  98     return rawTable.getScanTimeout(unit);
  99   }
 100
 101   private <T> CompletableFuture<T> wrap(CompletableFuture<T> future) {
 102     return FutureUtils.wrapFuture(future, pool);
 103   }
 104
 105   @Override
 106   public CompletableFuture<Result> get(Get get) {
 107     return wrap(rawTable.get(get));
 108   }
 109
 110   @Override
 111   public CompletableFuture<Void> put(Put put) {
 112     return wrap(rawTable.put(put));
 113   }
 114
 115   @Override
 116   public CompletableFuture<Void> delete(Delete delete) {
 117     return wrap(rawTable.delete(delete));
 118   }
 119
 120   @Override
 121   public CompletableFuture<Result> append(Append append) {
 122     return wrap(rawTable.append(append));
 123   }
 124
 125   @Override
 126   public CompletableFuture<Result> increment(Increment increment) {
 127     return wrap(rawTable.increment(increment));
 128   }
 129
 130   @Override
 131   public CheckAndMutateBuilder checkAndMutate(byte[] row, byte[] family) {
 132     return new CheckAndMutateBuilder() {
 133
 134       private final CheckAndMutateBuilder builder = rawTable.checkAndMutate(row, family);
 135
 136       @Override
 137       public CompletableFuture<Boolean> thenPut(Put put) {
 138         return wrap(builder.thenPut(put));
 139       }
 140
 141       @Override
 142       public CompletableFuture<Boolean> thenMutate(RowMutations mutation) {
 143         return wrap(builder.thenMutate(mutation));
 144       }
 145
 146       @Override
 147       public CompletableFuture<Boolean> thenDelete(Delete delete) {
 148         return wrap(builder.thenDelete(delete));
 149       }
 150
 151       @Override
 152       public CheckAndMutateBuilder qualifier(byte[] qualifier) {
 153         builder.qualifier(qualifier);
 154         return this;
 155       }
 156
 157       @Override
 158       public CheckAndMutateBuilder timeRange(TimeRange timeRange) {
 159         builder.timeRange(timeRange);
 160         return this;
 161       }
 162
 163       @Override
 164       public CheckAndMutateBuilder ifNotExists() {
 165         builder.ifNotExists();
 166         return this;
 167       }
 168
 169       @Override
 170       public CheckAndMutateBuilder ifMatches(CompareOperator compareOp, byte[] value) {
 171         builder.ifMatches(compareOp, value);
 172         return this;
 173       }
 174     };
 175   }
 176
 177   @Override
 178   public CheckAndMutateWithFilterBuilder checkAndMutate(byte[] row, Filter filter) {
 179     return new CheckAndMutateWithFilterBuilder() {
 180
 181       private final CheckAndMutateWithFilterBuilder builder =
 182         rawTable.checkAndMutate(row, filter);
 183
 184       @Override
 185       public CheckAndMutateWithFilterBuilder timeRange(TimeRange timeRange) {
 186         builder.timeRange(timeRange);
 187         return this;
 188       }
 189
 190       @Override
 191       public CompletableFuture<Boolean> thenPut(Put put) {
 192         return wrap(builder.thenPut(put));
 193       }
 194
 195       @Override
 196       public CompletableFuture<Boolean> thenDelete(Delete delete) {
 197         return wrap(builder.thenDelete(delete));
 198       }
 199
 200       @Override
 201       public CompletableFuture<Boolean> thenMutate(RowMutations mutation) {
 202         return wrap(builder.thenMutate(mutation));
 203       }
 204     };
 205   }
 206
 207   @Override
 208   public CompletableFuture<CheckAndMutateResult> checkAndMutate(CheckAndMutate checkAndMutate) {
 209     return wrap(rawTable.checkAndMutate(checkAndMutate));
 210   }
 211
 212   @Override
 213   public List<CompletableFuture<CheckAndMutateResult>> checkAndMutate(
 214     List<CheckAndMutate> checkAndMutates) {
 215     return rawTable.checkAndMutate(checkAndMutates).stream()
 216       .map(this::wrap).collect(toList());
 217   }
 218
 219   @Override
 220   public CompletableFuture<Result> mutateRow(RowMutations mutation) {
 221     return wrap(rawTable.mutateRow(mutation));
 222   }
 223
 224   @Override
 225   public CompletableFuture<List<Result>> scanAll(Scan scan) {
 226     return wrap(rawTable.scanAll(scan));
 227   }
 228
 229   @Override
 230   public ResultScanner getScanner(Scan scan) {
 231     return rawTable.getScanner(scan);
 232   }
 233
 234   private void scan0(Scan scan, ScanResultConsumer consumer) {
 235     try (ResultScanner scanner = getScanner(scan)) {
 236       consumer.onScanMetricsCreated(scanner.getScanMetrics());
 237       for (Result result; (result = scanner.next()) != null;) {
 238         if (!consumer.onNext(result)) {
 239           break;
 240         }
 241       }
 242       consumer.onComplete();
 243     } catch (IOException e) {
 244       consumer.onError(e);
 245     }
 246   }
 247
 248   @Override
 249   public void scan(Scan scan, ScanResultConsumer consumer) {
 250     pool.execute(() -> scan0(scan, consumer));
 251   }
 252
 253   @Override
 254   public List<CompletableFuture<Result>> get(List<Get> gets) {
 255     return rawTable.get(gets).stream().map(this::wrap).collect(toList());
 256   }
 257
 258   @Override
 259   public List<CompletableFuture<Void>> put(List<Put> puts) {
 260     return rawTable.put(puts).stream().map(this::wrap).collect(toList());
 261   }
 262
 263   @Override
 264   public List<CompletableFuture<Void>> delete(List<Delete> deletes) {
 265     return rawTable.delete(deletes).stream().map(this::wrap).collect(toList());
 266   }
 267
 268   @Override
 269   public <T> List<CompletableFuture<T>> batch(List<? extends Row> actions) {
 270     return rawTable.<T> batch(actions).stream().map(this::wrap).collect(toList());
 271   }
 272
 273   @Override
 274   public <S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
 275       ServiceCaller<S, R> callable, byte[] row) {
 276     return wrap(rawTable.coprocessorService(stubMaker, callable, row));
 277   }
 278
 279   @Override
 280   public <S, R> CoprocessorServiceBuilder<S, R> coprocessorService(
 281       Function<RpcChannel, S> stubMaker, ServiceCaller<S, R> callable,
 282       CoprocessorCallback<R> callback) {
 283     CoprocessorCallback<R> wrappedCallback = new CoprocessorCallback<R>() {
 284
 285       @Override
 286       public void onRegionComplete(RegionInfo region, R resp) {
 287         pool.execute(() -> callback.onRegionComplete(region, resp));
 288       }
 289
 290       @Override
 291       public void onRegionError(RegionInfo region, Throwable error) {
 292         pool.execute(() -> callback.onRegionError(region, error));
 293       }
 294
 295       @Override
 296       public void onComplete() {
 297         pool.execute(() -> callback.onComplete());
 298       }
 299
 300       @Override
 301       public void onError(Throwable error) {
 302         pool.execute(() -> callback.onError(error));
 303       }
 304     };
 305     CoprocessorServiceBuilder<S, R> builder =
 306       rawTable.coprocessorService(stubMaker, callable, wrappedCallback);
 307     return new CoprocessorServiceBuilder<S, R>() {
 308
 309       @Override
 310       public CoprocessorServiceBuilder<S, R> fromRow(byte[] startKey, boolean inclusive) {
 311         builder.fromRow(startKey, inclusive);
 312         return this;
 313       }
 314
 315       @Override
 316       public CoprocessorServiceBuilder<S, R> toRow(byte[] endKey, boolean inclusive) {
 317         builder.toRow(endKey, inclusive);
 318         return this;
 319       }
 320
 321       @Override
 322       public void execute() {
 323         builder.execute();
 324       }
 325     };
 326   }
 327 }