ovsdb/src/main/java/org/opendaylight/ovsdb/lib/OvsDBClientImpl.java

   1 /*
   2  * Copyright (C) 2014 EBay Software Foundation
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  *
   8  * Authors : Ashwin Raveendran
   9  */
  10 package org.opendaylight.ovsdb.lib;
  11
  12 import com.fasterxml.jackson.databind.JsonNode;
  13 import com.google.common.collect.Lists;
  14 import com.google.common.collect.Maps;
  15 import com.google.common.util.concurrent.FutureCallback;
  16 import com.google.common.util.concurrent.Futures;
  17 import com.google.common.util.concurrent.ListenableFuture;
  18 import com.google.common.util.concurrent.SettableFuture;
  19 import org.opendaylight.ovsdb.lib.message.MonitorRequest;
  20 import org.opendaylight.ovsdb.lib.message.OvsdbRPC;
  21 import org.opendaylight.ovsdb.lib.message.TransactBuilder;
  22 import org.opendaylight.ovsdb.lib.operations.Operation;
  23 import org.opendaylight.ovsdb.lib.operations.OperationResult;
  24 import org.opendaylight.ovsdb.lib.operations.TransactionBuilder;
  25 import org.opendaylight.ovsdb.lib.schema.DatabaseSchema;
  26
  27 import java.util.List;
  28 import java.util.Map;
  29 import java.util.Queue;
  30 import java.util.concurrent.ExecutorService;
  31
  32
  33 public class OvsDBClientImpl implements OvsDBClient {
  34
  35     ExecutorService executorService;
  36     String schemaName;
  37     OvsdbRPC rpc;
  38     Map<String, DatabaseSchema> schema = Maps.newHashMap();
  39     Queue<Throwable> exceptions;
  40
  41     public OvsDBClientImpl(OvsdbRPC rpc, ExecutorService executorService) {
  42         this.rpc = rpc;
  43         this.executorService = executorService;
  44     }
  45
  46     public OvsDBClientImpl() {
  47     }
  48
  49     @Override
  50     public ListenableFuture<List<OperationResult>> transact(List<Operation> operations) {
  51
  52         //todo, we may not need transactionbuilder if we can have JSON objects
  53         TransactBuilder builder = new TransactBuilder();
  54         for (Operation o : operations) {
  55             builder.addOperation(o);
  56         }
  57
  58         ListenableFuture<List<OperationResult>> transact = rpc.transact(builder);
  59         return transact;
  60     }
  61
  62     @Override
  63     public void monitor(MonitorRequest monitorRequest, MonitorCallBack callback) {
  64         throw new UnsupportedOperationException("not yet implemented");
  65     }
  66
  67     @Override
  68     public void cancelMonitor(MonitorHandle handler) {
  69         throw new UnsupportedOperationException("not yet implemented");
  70     }
  71
  72     @Override
  73     public void lock(String lockId, LockAquisitionCallback lockedCallBack, LockStolenCallback stolenCallback) {
  74         throw new UnsupportedOperationException("not yet implemented");
  75     }
  76
  77     @Override
  78     public ListenableFuture<Boolean> steal(String lockId) {
  79         throw new UnsupportedOperationException("not yet implemented");
  80     }
  81
  82     @Override
  83     public ListenableFuture<Boolean> unLock(String lockId) {
  84         throw new UnsupportedOperationException("not yet implemented");
  85     }
  86
  87     @Override
  88     public void startEchoService(EchoServiceCallbackFilters callbackFilters) {
  89         throw new UnsupportedOperationException("not yet implemented");
  90     }
  91
  92     @Override
  93     public void stopEchoService() {
  94         throw new UnsupportedOperationException("not yet implemented");
  95     }
  96
  97     @Override
  98     public TransactionBuilder transactBuilder() {
  99         return new TransactionBuilder(this);
 100     }
 101
 102
 103     public boolean isReady(long timeout) {
 104         //todo implement timeout
 105         return null != schema;
 106     }
 107
 108     @Override
 109     public ListenableFuture<List<String>> getDatabases() {
 110         return rpc.list_dbs();
 111     }
 112
 113     @Override
 114     public ListenableFuture<DatabaseSchema> getSchema(final String database, final boolean cacheResult) {
 115
 116         DatabaseSchema databaseSchema = schema.get(database);
 117
 118         if (databaseSchema == null) {
 119             ListenableFuture<Map<String, DatabaseSchema>> schemaFromDevice = getSchemaFromDevice(Lists.newArrayList(database));
 120
 121             final SettableFuture<DatabaseSchema> future = SettableFuture.create();
 122             Futures.addCallback(schemaFromDevice, new FutureCallback<Map<String, DatabaseSchema>>() {
 123                 @Override
 124                 public void onSuccess(Map<String, DatabaseSchema> result) {
 125                     if (result.containsKey(database)) {
 126                        DatabaseSchema s = result.get(database);
 127                        if (cacheResult) {
 128                          OvsDBClientImpl.this.schema.put(database, s);
 129                        }
 130                        future.set(s);
 131                     } else {
 132                         future.set(null);
 133                     }
 134                 }
 135
 136                 @Override
 137                 public void onFailure(Throwable t) {
 138                     //todo: should wrap
 139                     future.setException(t);
 140                 }
 141             });
 142           return future;
 143
 144         } else {
 145             return Futures.immediateFuture(databaseSchema);
 146         }
 147     }
 148
 149     private ListenableFuture<Map<String, DatabaseSchema>> getSchemaFromDevice(final List<String> dbNames) {
 150         Map<String, DatabaseSchema> schema = Maps.newHashMap();
 151         SettableFuture<Map<String, DatabaseSchema>> future = SettableFuture.create();
 152         _populateSchema(dbNames, schema, future);
 153         return future;
 154     }
 155
 156     private void _populateSchema(final List<String> dbNames,
 157                                                        final Map<String, DatabaseSchema> schema,
 158                                                        final SettableFuture<Map<String, DatabaseSchema>> sfuture) {
 159
 160         if (dbNames == null || dbNames.isEmpty()) {
 161             return;
 162         }
 163
 164         Futures.transform(rpc.get_schema(Lists.newArrayList(dbNames.get(0))),
 165                 new com.google.common.base.Function<JsonNode, Void>() {
 166             @Override
 167             public Void apply(JsonNode jsonNode) {
 168                 try{
 169                 schema.put(dbNames.get(0), DatabaseSchema.fromJson(jsonNode));
 170                 if (schema.size() > 1 && !sfuture.isCancelled()) {
 171                     _populateSchema(dbNames.subList(1, dbNames.size()), schema, sfuture);
 172                 } else if (schema.size() == 1) {
 173                     sfuture.set(schema);
 174                 }
 175             } catch (Throwable e) {
 176                sfuture.setException(e);
 177             }
 178             return null;
 179         }});
 180     }
 181
 182     public void setRpc(OvsdbRPC rpc) {
 183         this.rpc = rpc;
 184     }
 185
 186     public Queue<Throwable> getExceptions() {
 187         return exceptions;
 188     }
 189
 190 }