opendaylight/sal/yang-prototype/sal/sal-broker-impl/src/main/java/org/opendaylight/controller/sal/core/impl/data/DataBrokerModule.java

   1 /*\r
   2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.\r
   3  *\r
   4  * This program and the accompanying materials are made available under the\r
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,\r
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html\r
   7  */\r
   8 package org.opendaylight.controller.sal.core.impl.data;\r
   9 \r
  10 import java.util.ArrayList;\r
  11 import java.util.Collections;\r
  12 import java.util.HashSet;\r
  13 import java.util.List;\r
  14 import java.util.Map;\r
  15 import java.util.Set;\r
  16 import java.util.concurrent.Future;\r
  17 \r
  18 import org.opendaylight.controller.sal.common.DataStoreIdentifier;\r
  19 import org.opendaylight.controller.sal.core.api.BrokerService;\r
  20 import org.opendaylight.controller.sal.core.api.Broker.ConsumerSession;\r
  21 import org.opendaylight.controller.sal.core.api.Broker.ProviderSession;\r
  22 import org.opendaylight.controller.sal.core.api.Consumer.ConsumerFunctionality;\r
  23 import org.opendaylight.controller.sal.core.api.Provider.ProviderFunctionality;\r
  24 import org.opendaylight.controller.sal.core.api.data.DataBrokerService;\r
  25 import org.opendaylight.controller.sal.core.api.data.DataCommitHandler;\r
  26 import org.opendaylight.controller.sal.core.api.data.DataProviderService;\r
  27 import org.opendaylight.controller.sal.core.api.data.DataValidator;\r
  28 import org.opendaylight.controller.sal.core.api.data.DataCommitHandler.CommitTransaction;\r
  29 import org.opendaylight.controller.sal.core.api.data.DataProviderService.DataRefresher;\r
  30 import org.opendaylight.controller.sal.core.impl.RpcUtils;\r
  31 import org.opendaylight.controller.sal.core.spi.BrokerModule;\r
  32 import org.opendaylight.controller.yang.common.RpcError;\r
  33 import org.opendaylight.controller.yang.common.RpcResult;\r
  34 import org.opendaylight.controller.yang.data.api.CompositeNode;\r
  35 import org.opendaylight.controller.yang.data.api.CompositeNodeModification;\r
  36 import org.slf4j.Logger;\r
  37 import org.slf4j.LoggerFactory;\r
  38 \r
  39 \r
  40 public class DataBrokerModule implements BrokerModule {\r
  41 \r
  42     private static final Logger log = LoggerFactory\r
  43             .getLogger(DataBrokerModule.class);\r
  44 \r
  45     private Map<DataStoreIdentifier, StoreContext> storeContext;\r
  46 \r
  47     private SequentialCommitHandlerCoordinator coordinator = new SequentialCommitHandlerCoordinator();\r
  48 \r
  49     @Override\r
  50     public Set<Class<? extends BrokerService>> getProvidedServices() {\r
  51         // FIXME: Refactor\r
  52         Set<Class<? extends BrokerService>> ret = new HashSet<Class<? extends BrokerService>>();\r
  53         ret.add(DataBrokerService.class);\r
  54         ret.add(DataProviderService.class);\r
  55         return ret;\r
  56     }\r
  57 \r
  58     @Override\r
  59     public Set<Class<? extends ProviderFunctionality>> getSupportedProviderFunctionality() {\r
  60         // FIXME Refactor\r
  61         Set<Class<? extends ProviderFunctionality>> ret = new HashSet<Class<? extends ProviderFunctionality>>();\r
  62         ret.add(DataValidator.class);\r
  63         ret.add(DataCommitHandler.class);\r
  64         ret.add(DataRefresher.class);\r
  65         return ret;\r
  66     }\r
  67 \r
  68     @Override\r
  69     public <T extends BrokerService> T getServiceForSession(Class<T> service,\r
  70             ConsumerSession session) {\r
  71         if (DataProviderService.class.equals(service)\r
  72                 && session instanceof ProviderSession) {\r
  73             @SuppressWarnings("unchecked")\r
  74             T ret = (T) newDataProviderService(session);\r
  75             return ret;\r
  76         } else if (DataBrokerService.class.equals(service)) {\r
  77 \r
  78             @SuppressWarnings("unchecked")\r
  79             T ret = (T) newDataConsumerService(session);\r
  80             return ret;\r
  81         }\r
  82 \r
  83         throw new IllegalArgumentException(\r
  84                 "The requested session-specific service is not provided by this module.");\r
  85     }\r
  86 \r
  87     private DataProviderService newDataProviderService(ConsumerSession session) {\r
  88         // TODO Implement this method\r
  89         throw new UnsupportedOperationException("Not implemented");\r
  90     }\r
  91 \r
  92     private DataBrokerService newDataConsumerService(ConsumerSession session) {\r
  93         // TODO Implement this method\r
  94         throw new UnsupportedOperationException("Not implemented");\r
  95     }\r
  96 \r
  97     @Override\r
  98     public Set<Class<? extends ConsumerFunctionality>> getSupportedConsumerFunctionality() {\r
  99         // TODO Implement this method\r
 100         throw new UnsupportedOperationException("Not implemented");\r
 101     }\r
 102 \r
 103     private static class StoreContext {\r
 104         private Set<DataCommitHandler> commitHandlers = new HashSet<DataCommitHandler>();\r
 105         private Set<DataValidator> validators = new HashSet<DataValidator>();\r
 106         private Set<DataRefresher> refreshers = new HashSet<DataRefresher>();\r
 107     }\r
 108 \r
 109     private class DataConsumerSession implements DataBrokerService {\r
 110 \r
 111         @Override\r
 112         public CompositeNode getData(DataStoreIdentifier store) {\r
 113             // TODO Implement this method\r
 114             throw new UnsupportedOperationException("Not implemented");\r
 115         }\r
 116 \r
 117         @Override\r
 118         public CompositeNode getData(DataStoreIdentifier store,\r
 119                 CompositeNode filter) {\r
 120             // TODO Implement this method\r
 121             throw new UnsupportedOperationException("Not implemented");\r
 122         }\r
 123 \r
 124         @Override\r
 125         public CompositeNode getCandidateData(DataStoreIdentifier store) {\r
 126             // TODO Implement this method\r
 127             throw new UnsupportedOperationException("Not implemented");\r
 128         }\r
 129 \r
 130         @Override\r
 131         public CompositeNode getCandidateData(DataStoreIdentifier store,\r
 132                 CompositeNode filter) {\r
 133             // TODO Implement this method\r
 134             throw new UnsupportedOperationException("Not implemented");\r
 135         }\r
 136 \r
 137         @Override\r
 138         public RpcResult<CompositeNode> editCandidateData(\r
 139                 DataStoreIdentifier store, CompositeNodeModification changeSet) {\r
 140             // TODO Implement this method\r
 141             throw new UnsupportedOperationException("Not implemented");\r
 142         }\r
 143 \r
 144         @Override\r
 145         public Future<RpcResult<Void>> commit(DataStoreIdentifier store) {\r
 146             // TODO Implement this method\r
 147             throw new UnsupportedOperationException("Not implemented");\r
 148         }\r
 149 \r
 150         @Override\r
 151         public void closeSession() {\r
 152             // TODO Implement this method\r
 153             throw new UnsupportedOperationException("Not implemented");\r
 154         }\r
 155 \r
 156     }\r
 157 \r
 158     private StoreContext context(DataStoreIdentifier store) {\r
 159         return storeContext.get(store);\r
 160     }\r
 161 \r
 162     private class DataProviderSession extends DataConsumerSession implements\r
 163             DataProviderService {\r
 164 \r
 165         private Set<DataCommitHandler> providerCommitHandlers = new HashSet<DataCommitHandler>();\r
 166         private Set<DataValidator> providerValidators = new HashSet<DataValidator>();\r
 167         private Set<DataRefresher> providerRefreshers = new HashSet<DataRefresher>();\r
 168 \r
 169         @Override\r
 170         public void addValidator(DataStoreIdentifier store,\r
 171                 DataValidator validator) {\r
 172             if (validator == null)\r
 173                 throw new IllegalArgumentException(\r
 174                         "Validator should not be null");\r
 175 \r
 176             providerValidators.add(validator);\r
 177             context(store).validators.add(validator);\r
 178         }\r
 179 \r
 180         @Override\r
 181         public void removeValidator(DataStoreIdentifier store,\r
 182                 DataValidator validator) {\r
 183             if (validator == null)\r
 184                 throw new IllegalArgumentException(\r
 185                         "Validator should not be null");\r
 186 \r
 187             providerValidators.remove(validator);\r
 188             context(store).validators.remove(validator);\r
 189         }\r
 190 \r
 191         @Override\r
 192         public void addCommitHandler(DataStoreIdentifier store,\r
 193                 DataCommitHandler provider) {\r
 194             if (provider == null)\r
 195                 throw new IllegalArgumentException(\r
 196                         "CommitHandler should not be null");\r
 197 \r
 198             providerCommitHandlers.add(provider);\r
 199             context(store).commitHandlers.add(provider);\r
 200         }\r
 201 \r
 202         @Override\r
 203         public void removeCommitHandler(DataStoreIdentifier store,\r
 204                 DataCommitHandler provider) {\r
 205             if (provider == null)\r
 206                 throw new IllegalArgumentException(\r
 207                         "CommitHandler should not be null");\r
 208 \r
 209             providerCommitHandlers.remove(provider);\r
 210             context(store).commitHandlers.remove(provider);\r
 211         }\r
 212 \r
 213         @Override\r
 214         public void addRefresher(DataStoreIdentifier store,\r
 215                 DataRefresher provider) {\r
 216             if (provider == null)\r
 217                 throw new IllegalArgumentException(\r
 218                         "Refresher should not be null");\r
 219 \r
 220             providerRefreshers.add(provider);\r
 221             context(store).refreshers.add(provider);\r
 222         }\r
 223 \r
 224         @Override\r
 225         public void removeRefresher(DataStoreIdentifier store,\r
 226                 DataRefresher provider) {\r
 227             if (provider == null)\r
 228                 throw new IllegalArgumentException(\r
 229                         "Refresher should not be null");\r
 230 \r
 231             providerRefreshers.remove(provider);\r
 232             context(store).refreshers.remove(provider);\r
 233         }\r
 234 \r
 235     }\r
 236 \r
 237     private class SequentialCommitHandlerCoordinator implements\r
 238             DataCommitHandler {\r
 239 \r
 240         @Override\r
 241         public RpcResult<CommitTransaction> requestCommit(\r
 242                 DataStoreIdentifier store) {\r
 243             List<RpcError> errors = new ArrayList<RpcError>();\r
 244             Set<CommitTransaction> transactions = new HashSet<DataCommitHandler.CommitTransaction>();\r
 245             boolean successful = true;\r
 246 \r
 247             for (DataCommitHandler commitHandler : context(store).commitHandlers) {\r
 248                 try {\r
 249                     RpcResult<CommitTransaction> partialResult = commitHandler\r
 250                             .requestCommit(store);\r
 251                     successful = partialResult.isSuccessful() & successful;\r
 252                     if (partialResult.isSuccessful()) {\r
 253                         transactions.add(partialResult.getResult());\r
 254                     }\r
 255 \r
 256                     errors.addAll(partialResult.getErrors());\r
 257                 } catch (Exception e) {\r
 258                     log.error("Uncaught exception prevented commit request."\r
 259                             + e.getMessage(), e);\r
 260                     successful = false;\r
 261                     // FIXME: Add RPC Error with exception.\r
 262                 }\r
 263                 if (successful == false)\r
 264                     break;\r
 265             }\r
 266             CommitTransaction transaction = new SequentialCommitTransaction(\r
 267                     store, transactions);\r
 268             return RpcUtils.getRpcResult(successful, transaction, errors);\r
 269         }\r
 270 \r
 271         @Override\r
 272         public Set<DataStoreIdentifier> getSupportedDataStores() {\r
 273             return Collections.emptySet();\r
 274         }\r
 275     }\r
 276 \r
 277     private class SequentialCommitTransaction implements CommitTransaction {\r
 278 \r
 279         final Set<CommitTransaction> transactions;\r
 280         final DataStoreIdentifier store;\r
 281 \r
 282         public SequentialCommitTransaction(DataStoreIdentifier s,\r
 283                 Set<CommitTransaction> t) {\r
 284             transactions = t;\r
 285             store = s;\r
 286         }\r
 287 \r
 288         @Override\r
 289         public RpcResult<Void> finish() {\r
 290             List<RpcError> errors = new ArrayList<RpcError>();\r
 291             boolean successful = true;\r
 292 \r
 293             for (CommitTransaction commitHandler : transactions) {\r
 294                 try {\r
 295                     RpcResult<Void> partialResult = commitHandler.finish();\r
 296                     successful = partialResult.isSuccessful() & successful;\r
 297                     errors.addAll(partialResult.getErrors());\r
 298                 } catch (Exception e) {\r
 299                     log.error(\r
 300                             "Uncaught exception prevented finishing of commit."\r
 301                                     + e.getMessage(), e);\r
 302                     successful = false;\r
 303                     // FIXME: Add RPC Error with exception.\r
 304                 }\r
 305                 if (successful == false)\r
 306                     break;\r
 307             }\r
 308 \r
 309             return RpcUtils.getRpcResult(successful, null, errors);\r
 310         }\r
 311 \r
 312         @Override\r
 313         public RpcResult<Void> rollback() {\r
 314             List<RpcError> errors = new ArrayList<RpcError>();\r
 315             boolean successful = true;\r
 316 \r
 317             for (CommitTransaction commitHandler : transactions) {\r
 318                 try {\r
 319                     RpcResult<Void> partialResult = commitHandler.rollback();\r
 320                     successful = partialResult.isSuccessful() & successful;\r
 321                     errors.addAll(partialResult.getErrors());\r
 322                 } catch (Exception e) {\r
 323                     log.error(\r
 324                             "Uncaught exception prevented rollback of commit."\r
 325                                     + e.getMessage(), e);\r
 326                     successful = false;\r
 327                     // FIXME: Add RPC Error with exception.\r
 328                 }\r
 329                 if (successful == false)\r
 330                     break;\r
 331             }\r
 332 \r
 333             return RpcUtils.getRpcResult(successful, null, errors);\r
 334         }\r
 335 \r
 336         @Override\r
 337         public DataStoreIdentifier getDataStore() {\r
 338             return this.store;\r
 339         }\r
 340 \r
 341         @Override\r
 342         public DataCommitHandler getHandler() {\r
 343             return coordinator;\r
 344         }\r
 345     }\r
 346 \r
 347     private class ValidationCoordinator implements DataValidator {\r
 348 \r
 349         private final DataStoreIdentifier store;\r
 350 \r
 351         ValidationCoordinator(DataStoreIdentifier store) {\r
 352             this.store = store;\r
 353         }\r
 354 \r
 355         @Override\r
 356         public RpcResult<Void> validate(CompositeNode toValidate) {\r
 357             List<RpcError> errors = new ArrayList<RpcError>();\r
 358             boolean successful = true;\r
 359 \r
 360             for (DataValidator validator : context(store).validators) {\r
 361                 try {\r
 362                     RpcResult<Void> partialResult = validator\r
 363                             .validate(toValidate);\r
 364                     successful = partialResult.isSuccessful() & successful;\r
 365                     errors.addAll(partialResult.getErrors());\r
 366                 } catch (Exception e) {\r
 367                     log.error(\r
 368                             "Uncaught exception prevented validation."\r
 369                                     + e.getMessage(), e);\r
 370                     successful = false;\r
 371                     // FIXME: Add RPC Error with exception.\r
 372                 }\r
 373                 if (successful == false)\r
 374                     break;\r
 375             }\r
 376 \r
 377             return RpcUtils.getRpcResult(successful, null, errors);\r
 378         }\r
 379 \r
 380         @Override\r
 381         public Set<DataStoreIdentifier> getSupportedDataStores() {\r
 382             return Collections.emptySet();\r
 383         }\r
 384 \r
 385     }\r
 386 \r
 387     private class DataRefreshCoordinator implements DataRefresher {\r
 388 \r
 389         private final DataStoreIdentifier store;\r
 390 \r
 391         DataRefreshCoordinator(DataStoreIdentifier store) {\r
 392             this.store = store;\r
 393         }\r
 394 \r
 395         @Override\r
 396         public void refreshData() {\r
 397 \r
 398             for (DataRefresher refresher : context(store).refreshers) {\r
 399                 try {\r
 400                     refresher.refreshData();\r
 401                 } catch (Exception e) {\r
 402                     log.error(\r
 403                             "Uncaught exception during refresh of data: "\r
 404                                     + e.getMessage(), e);\r
 405                 }\r
 406 \r
 407             }\r
 408         }\r
 409     }\r
 410 }\r
 411