]> git.basschouten.com Git - openhab-addons.git/blob
c3426c513d092bdae328787d9b25ec54ecee2845
[openhab-addons.git] /
1 /**
2  * Copyright (c) 2010-2022 Contributors to the openHAB project
3  *
4  * See the NOTICE file(s) distributed with this work for additional
5  * information.
6  *
7  * This program and the accompanying materials are made available under the
8  * terms of the Eclipse Public License 2.0 which is available at
9  * http://www.eclipse.org/legal/epl-2.0
10  *
11  * SPDX-License-Identifier: EPL-2.0
12  */
13 package org.openhab.persistence.dynamodb.internal;
14
15 import java.lang.reflect.InvocationTargetException;
16 import java.net.URI;
17 import java.time.Duration;
18 import java.time.Instant;
19 import java.time.ZonedDateTime;
20 import java.util.Collections;
21 import java.util.Iterator;
22 import java.util.List;
23 import java.util.Locale;
24 import java.util.Map;
25 import java.util.Set;
26 import java.util.concurrent.CompletableFuture;
27 import java.util.concurrent.ConcurrentHashMap;
28 import java.util.concurrent.ExecutionException;
29 import java.util.concurrent.ExecutorService;
30 import java.util.concurrent.ThreadPoolExecutor;
31 import java.util.stream.Collectors;
32
33 import javax.measure.Unit;
34
35 import org.eclipse.jdt.annotation.NonNullByDefault;
36 import org.eclipse.jdt.annotation.Nullable;
37 import org.openhab.core.common.ThreadPoolManager;
38 import org.openhab.core.config.core.ConfigurableService;
39 import org.openhab.core.items.GenericItem;
40 import org.openhab.core.items.GroupItem;
41 import org.openhab.core.items.Item;
42 import org.openhab.core.items.ItemNotFoundException;
43 import org.openhab.core.items.ItemRegistry;
44 import org.openhab.core.library.items.NumberItem;
45 import org.openhab.core.library.types.QuantityType;
46 import org.openhab.core.persistence.FilterCriteria;
47 import org.openhab.core.persistence.HistoricItem;
48 import org.openhab.core.persistence.PersistenceItemInfo;
49 import org.openhab.core.persistence.PersistenceService;
50 import org.openhab.core.persistence.QueryablePersistenceService;
51 import org.openhab.core.persistence.strategy.PersistenceStrategy;
52 import org.openhab.core.types.State;
53 import org.openhab.core.types.UnDefType;
54 import org.osgi.framework.BundleContext;
55 import org.osgi.framework.Constants;
56 import org.osgi.service.component.annotations.Activate;
57 import org.osgi.service.component.annotations.Component;
58 import org.osgi.service.component.annotations.Deactivate;
59 import org.osgi.service.component.annotations.Reference;
60 import org.reactivestreams.Subscriber;
61 import org.slf4j.Logger;
62 import org.slf4j.LoggerFactory;
63
64 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
65 import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration;
66 import software.amazon.awssdk.awscore.defaultsmode.DefaultsMode;
67 import software.amazon.awssdk.core.async.SdkPublisher;
68 import software.amazon.awssdk.core.client.config.ClientAsyncConfiguration;
69 import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
70 import software.amazon.awssdk.core.client.config.SdkAdvancedAsyncClientOption;
71 import software.amazon.awssdk.enhanced.dynamodb.DynamoDbAsyncTable;
72 import software.amazon.awssdk.enhanced.dynamodb.DynamoDbEnhancedAsyncClient;
73 import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
74 import software.amazon.awssdk.enhanced.dynamodb.model.QueryEnhancedRequest;
75 import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
76 import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
77 import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClientBuilder;
78 import software.amazon.awssdk.services.dynamodb.model.ResourceNotFoundException;
79
80 /**
81  * This is the implementation of the DynamoDB {@link PersistenceService}. It persists item values
82  * using the <a href="https://aws.amazon.com/dynamodb/">Amazon DynamoDB</a> database. The states (
83  * {@link State}) of an {@link Item} are persisted in DynamoDB tables.
84  *
85  * The service creates tables automatically, one for numbers, and one for strings.
86  *
87  * @see AbstractDynamoDBItem.fromState for details how different items are persisted
88  *
89  * @author Sami Salonen - Initial contribution
90  * @author Kai Kreuzer - Migration to 3.x
91  *
92  */
93 @NonNullByDefault
94 @Component(service = { PersistenceService.class,
95         QueryablePersistenceService.class }, configurationPid = "org.openhab.dynamodb", //
96         property = Constants.SERVICE_PID + "=org.openhab.dynamodb")
97 @ConfigurableService(category = "persistence", label = "DynamoDB Persistence Service", description_uri = DynamoDBPersistenceService.CONFIG_URI)
98 public class DynamoDBPersistenceService implements QueryablePersistenceService {
99
100     private static final int MAX_CONCURRENCY = 100;
101
102     protected static final String CONFIG_URI = "persistence:dynamodb";
103
104     private static final String DYNAMODB_THREADPOOL_NAME = "dynamodbPersistenceService";
105
106     private ItemRegistry itemRegistry;
107     private @Nullable DynamoDbEnhancedAsyncClient client;
108     private @Nullable DynamoDbAsyncClient lowLevelClient;
109     private final static Logger logger = LoggerFactory.getLogger(DynamoDBPersistenceService.class);
110     private boolean isProperlyConfigured;
111     private @Nullable DynamoDBConfig dbConfig;
112     private @Nullable DynamoDBTableNameResolver tableNameResolver;
113     private final ExecutorService executor = ThreadPoolManager.getPool(DYNAMODB_THREADPOOL_NAME);
114     private static final Duration TIMEOUT_API_CALL = Duration.ofSeconds(60);
115     private static final Duration TIMEOUT_API_CALL_ATTEMPT = Duration.ofSeconds(5);
116     private Map<Class<? extends DynamoDBItem<?>>, DynamoDbAsyncTable<? extends DynamoDBItem<?>>> tableCache = new ConcurrentHashMap<>(
117             2);
118
119     private @Nullable URI endpointOverride;
120
121     void overrideConfig(AwsRequestOverrideConfiguration.Builder config) {
122         config.apiCallAttemptTimeout(TIMEOUT_API_CALL_ATTEMPT).apiCallTimeout(TIMEOUT_API_CALL);
123     }
124
125     void overrideConfig(ClientOverrideConfiguration.Builder config) {
126         DynamoDBConfig localDbConfig = dbConfig;
127         config.apiCallAttemptTimeout(TIMEOUT_API_CALL_ATTEMPT).apiCallTimeout(TIMEOUT_API_CALL);
128         if (localDbConfig != null) {
129             localDbConfig.getRetryPolicy().ifPresent(config::retryPolicy);
130         }
131     }
132
133     @Activate
134     public DynamoDBPersistenceService(final @Reference ItemRegistry itemRegistry) {
135         this.itemRegistry = itemRegistry;
136     }
137
138     /**
139      * For tests
140      */
141     DynamoDBPersistenceService(final ItemRegistry itemRegistry, @Nullable URI endpointOverride) {
142         this.itemRegistry = itemRegistry;
143         this.endpointOverride = endpointOverride;
144     }
145
146     /**
147      * For tests
148      */
149     @Nullable
150     URI getEndpointOverride() {
151         return endpointOverride;
152     }
153
154     @Nullable
155     DynamoDbAsyncClient getLowLevelClient() {
156         return lowLevelClient;
157     }
158
159     ExecutorService getExecutor() {
160         return executor;
161     }
162
163     @Nullable
164     DynamoDBTableNameResolver getTableNameResolver() {
165         return tableNameResolver;
166     }
167
168     @Nullable
169     DynamoDBConfig getDbConfig() {
170         return dbConfig;
171     }
172
173     @Activate
174     public void activate(final @Nullable BundleContext bundleContext, final Map<String, Object> config) {
175         disconnect();
176         DynamoDBConfig localDbConfig = dbConfig = DynamoDBConfig.fromConfig(config);
177         if (localDbConfig == null) {
178             // Configuration was invalid. Abort service activation.
179             // Error is already logger in fromConfig.
180             return;
181         }
182         tableNameResolver = new DynamoDBTableNameResolver(localDbConfig.getTableRevision(), localDbConfig.getTable(),
183                 localDbConfig.getTablePrefixLegacy());
184         try {
185             if (!ensureClient()) {
186                 logger.error("Error creating dynamodb database client. Aborting service activation.");
187                 return;
188             }
189         } catch (Exception e) {
190             logger.error("Error constructing dynamodb client", e);
191             return;
192         }
193
194         isProperlyConfigured = true;
195         logger.debug("dynamodb persistence service activated");
196     }
197
198     @Deactivate
199     public void deactivate() {
200         logger.debug("dynamodb persistence service deactivated");
201         logIfManyQueuedTasks();
202         disconnect();
203     }
204
205     /**
206      * Initializes Dynamo DB client and determines schema
207      *
208      * If construction fails, error is logged and false is returned.
209      *
210      * @return whether initialization was successful.
211      */
212     private boolean ensureClient() {
213         DynamoDBConfig localDbConfig = dbConfig;
214         if (localDbConfig == null) {
215             return false;
216         }
217         if (client == null) {
218             try {
219                 synchronized (this) {
220                     if (this.client != null) {
221                         return true;
222                     }
223                     DynamoDbAsyncClientBuilder lowlevelClientBuilder = DynamoDbAsyncClient.builder()
224                             .defaultsMode(DefaultsMode.STANDARD)
225                             .credentialsProvider(StaticCredentialsProvider.create(localDbConfig.getCredentials()))
226                             .httpClient(NettyNioAsyncHttpClient.builder().maxConcurrency(MAX_CONCURRENCY).build())
227                             .asyncConfiguration(
228                                     ClientAsyncConfiguration.builder()
229                                             .advancedOption(SdkAdvancedAsyncClientOption.FUTURE_COMPLETION_EXECUTOR,
230                                                     executor)
231                                             .build())
232                             .overrideConfiguration(this::overrideConfig).region(localDbConfig.getRegion());
233                     if (endpointOverride != null) {
234                         logger.debug("DynamoDB has been overriden to {}", endpointOverride);
235                         lowlevelClientBuilder.endpointOverride(endpointOverride);
236                     }
237                     DynamoDbAsyncClient lowlevelClient = lowlevelClientBuilder.build();
238                     client = DynamoDbEnhancedAsyncClient.builder().dynamoDbClient(lowlevelClient).build();
239                     this.lowLevelClient = lowlevelClient;
240                 }
241             } catch (Exception e) {
242                 logger.error("Error constructing dynamodb client", e);
243                 return false;
244             }
245         }
246         return true;
247     }
248
249     private CompletableFuture<Boolean> resolveTableSchema() {
250         DynamoDBTableNameResolver localTableNameResolver = tableNameResolver;
251         DynamoDbAsyncClient localLowLevelClient = lowLevelClient;
252         if (localTableNameResolver == null || localLowLevelClient == null) {
253             throw new IllegalStateException("tableNameResolver or localLowLevelClient not available");
254         }
255         if (localTableNameResolver.isFullyResolved()) {
256             return CompletableFuture.completedFuture(true);
257         } else {
258             synchronized (localTableNameResolver) {
259                 if (localTableNameResolver.isFullyResolved()) {
260                     return CompletableFuture.completedFuture(true);
261                 }
262                 return localTableNameResolver.resolveSchema(localLowLevelClient,
263                         b -> b.overrideConfiguration(this::overrideConfig), executor).thenApplyAsync(resolved -> {
264                             if (resolved && localTableNameResolver.getTableSchema() == ExpectedTableSchema.LEGACY) {
265                                 logger.warn(
266                                         "Using legacy table format. Is it recommended to migrate to the new table format: specify the 'table' parameter and unset the old 'tablePrefix' parameter.");
267                             }
268                             return resolved;
269                         }, executor);
270             }
271         }
272     }
273
274     private <T extends DynamoDBItem<?>> DynamoDbAsyncTable<T> getTable(Class<T> dtoClass) {
275         DynamoDbEnhancedAsyncClient localClient = client;
276         DynamoDBTableNameResolver localTableNameResolver = tableNameResolver;
277         if (!ensureClient() || localClient == null || localTableNameResolver == null) {
278             throw new IllegalStateException("Client not ready");
279         }
280         ExpectedTableSchema expectedTableSchemaRevision = localTableNameResolver.getTableSchema();
281         String tableName = localTableNameResolver.fromClass(dtoClass);
282         final TableSchema<T> schema = getDynamoDBTableSchema(dtoClass, expectedTableSchemaRevision);
283         @SuppressWarnings("unchecked") // OK since this is the only place tableCache is populated
284         DynamoDbAsyncTable<T> table = (DynamoDbAsyncTable<T>) tableCache.computeIfAbsent(dtoClass, clz -> {
285             return localClient.table(tableName, schema);
286         });
287         if (table == null) {
288             // Invariant. To make null checker happy
289             throw new IllegalStateException();
290         }
291         return table;
292     }
293
294     private static <T extends DynamoDBItem<?>> TableSchema<T> getDynamoDBTableSchema(Class<T> dtoClass,
295             ExpectedTableSchema expectedTableSchemaRevision) {
296         if (dtoClass.equals(DynamoDBBigDecimalItem.class)) {
297             @SuppressWarnings("unchecked") // OK thanks to above conditional
298             TableSchema<T> schema = (TableSchema<T>) (expectedTableSchemaRevision == ExpectedTableSchema.NEW
299                     ? DynamoDBBigDecimalItem.TABLE_SCHEMA_NEW
300                     : DynamoDBBigDecimalItem.TABLE_SCHEMA_LEGACY);
301             return schema;
302         } else if (dtoClass.equals(DynamoDBStringItem.class)) {
303             @SuppressWarnings("unchecked") // OK thanks to above conditional
304             TableSchema<T> schema = (TableSchema<T>) (expectedTableSchemaRevision == ExpectedTableSchema.NEW
305                     ? DynamoDBStringItem.TABLE_SCHEMA_NEW
306                     : DynamoDBStringItem.TABLE_SCHEMA_LEGACY);
307             return schema;
308         } else {
309             throw new IllegalStateException("Unknown DTO class. Bug");
310         }
311     }
312
313     private void disconnect() {
314         DynamoDbAsyncClient localLowLevelClient = lowLevelClient;
315         if (client == null || localLowLevelClient == null) {
316             return;
317         }
318         localLowLevelClient.close();
319         lowLevelClient = null;
320         client = null;
321         dbConfig = null;
322         tableNameResolver = null;
323         isProperlyConfigured = false;
324         tableCache.clear();
325     }
326
327     protected boolean isReadyToStore() {
328         return isProperlyConfigured && ensureClient();
329     }
330
331     @Override
332     public String getId() {
333         return "dynamodb";
334     }
335
336     @Override
337     public String getLabel(@Nullable Locale locale) {
338         return "DynamoDB";
339     }
340
341     @Override
342     public Set<PersistenceItemInfo> getItemInfo() {
343         return Collections.emptySet();
344     }
345
346     @Override
347     public Iterable<HistoricItem> query(FilterCriteria filter) {
348         logIfManyQueuedTasks();
349         Instant start = Instant.now();
350         String filterDescription = filterToString(filter);
351         logger.trace("Got a query with filter {}", filterDescription);
352         DynamoDbEnhancedAsyncClient localClient = client;
353         DynamoDBTableNameResolver localTableNameResolver = tableNameResolver;
354         if (!isProperlyConfigured) {
355             logger.debug("Configuration for dynamodb not yet loaded or broken. Returning empty query results.");
356             return Collections.emptyList();
357         }
358         if (!ensureClient() || localClient == null || localTableNameResolver == null) {
359             logger.warn("DynamoDB not connected. Returning empty query results.");
360             return Collections.emptyList();
361         }
362
363         //
364         // Resolve unclear table schema if needed
365         //
366         try {
367             Boolean resolved = resolveTableSchema().get();
368             if (!resolved) {
369                 logger.warn("Table schema not resolved, cannot query data.");
370                 return Collections.emptyList();
371             }
372         } catch (InterruptedException e) {
373             logger.warn("Table schema resolution interrupted, cannot query data");
374             return Collections.emptyList();
375         } catch (ExecutionException e) {
376             Throwable cause = e.getCause();
377             logger.warn("Table schema resolution errored, cannot query data: {} {}",
378                     cause == null ? e.getClass().getSimpleName() : cause.getClass().getSimpleName(),
379                     cause == null ? e.getMessage() : cause.getMessage());
380             return Collections.emptyList();
381         }
382         try {
383             //
384             // Proceed with query
385             //
386             String itemName = filter.getItemName();
387             Item item = getItemFromRegistry(itemName);
388             if (item == null) {
389                 logger.warn("Could not get item {} from registry! Returning empty query results.", itemName);
390                 return Collections.emptyList();
391             }
392             if (item instanceof GroupItem) {
393                 item = ((GroupItem) item).getBaseItem();
394                 logger.debug("Item is instanceof GroupItem '{}'", itemName);
395                 if (item == null) {
396                     logger.debug("BaseItem of GroupItem is null. Ignore and give up!");
397                     return Collections.emptyList();
398                 }
399                 if (item instanceof GroupItem) {
400                     logger.debug("BaseItem of GroupItem is a GroupItem too. Ignore and give up!");
401                     return Collections.emptyList();
402                 }
403             }
404             boolean legacy = localTableNameResolver.getTableSchema() == ExpectedTableSchema.LEGACY;
405             Class<? extends DynamoDBItem<?>> dtoClass = AbstractDynamoDBItem.getDynamoItemClass(item.getClass(),
406                     legacy);
407             String tableName = localTableNameResolver.fromClass(dtoClass);
408             DynamoDbAsyncTable<? extends DynamoDBItem<?>> table = getTable(dtoClass);
409             logger.debug("Item {} (of type {}) will be tried to query using DTO class {} from table {}", itemName,
410                     item.getClass().getSimpleName(), dtoClass.getSimpleName(), tableName);
411
412             QueryEnhancedRequest queryExpression = DynamoDBQueryUtils.createQueryExpression(dtoClass,
413                     localTableNameResolver.getTableSchema(), item, filter);
414
415             CompletableFuture<List<DynamoDBItem<?>>> itemsFuture = new CompletableFuture<>();
416             final SdkPublisher<? extends DynamoDBItem<?>> itemPublisher = table.query(queryExpression).items();
417             Subscriber<DynamoDBItem<?>> pageSubscriber = new PageOfInterestSubscriber<DynamoDBItem<?>>(itemsFuture,
418                     filter.getPageNumber(), filter.getPageSize());
419             itemPublisher.subscribe(pageSubscriber);
420             // NumberItem.getUnit() is expensive, we avoid calling it in the loop
421             // by fetching the unit here.
422             final Item localItem = item;
423             final Unit<?> itemUnit = localItem instanceof NumberItem ? ((NumberItem) localItem).getUnit() : null;
424             try {
425                 @SuppressWarnings("null")
426                 List<HistoricItem> results = itemsFuture.get().stream().map(dynamoItem -> {
427                     HistoricItem historicItem = dynamoItem.asHistoricItem(localItem, itemUnit);
428                     if (historicItem == null) {
429                         logger.warn(
430                                 "Dynamo item {} serialized state '{}' cannot be converted to item {} {}. Item type changed since persistence. Ignoring",
431                                 dynamoItem.getClass().getSimpleName(), dynamoItem.getState(),
432                                 localItem.getClass().getSimpleName(), localItem.getName());
433                         return null;
434                     }
435                     logger.trace("Dynamo item {} converted to historic item: {}", localItem, historicItem);
436                     return historicItem;
437                 }).filter(value -> value != null).collect(Collectors.toList());
438                 logger.debug("Query completed in {} ms. Filter was {}",
439                         Duration.between(start, Instant.now()).toMillis(), filterDescription);
440                 return results;
441             } catch (InterruptedException e) {
442                 logger.warn("Query interrupted. Filter was {}", filterDescription);
443                 return Collections.emptyList();
444             } catch (ExecutionException e) {
445                 Throwable cause = e.getCause();
446                 if (cause instanceof ResourceNotFoundException) {
447                     logger.trace("Query failed since the DynamoDB table '{}' does not exist. Filter was {}", tableName,
448                             filterDescription);
449                 } else if (logger.isTraceEnabled()) {
450                     logger.trace("Query failed. Filter was {}", filterDescription, e);
451                 } else {
452                     logger.warn("Query failed {} {}. Filter was {}",
453                             cause == null ? e.getClass().getSimpleName() : cause.getClass().getSimpleName(),
454                             cause == null ? e.getMessage() : cause.getMessage(), filterDescription);
455                 }
456                 return Collections.emptyList();
457             }
458         } catch (Exception e) {
459             logger.error("Unexpected error with query having filter {}: {} {}. Returning empty query results.",
460                     filterDescription, e.getClass().getSimpleName(), e.getMessage());
461             return Collections.emptyList();
462         }
463     }
464
465     /**
466      * Retrieves the item for the given name from the item registry
467      *
468      * @param itemName
469      * @return item with the given name, or null if no such item exists in item registry.
470      */
471     private @Nullable Item getItemFromRegistry(String itemName) {
472         try {
473             return itemRegistry.getItem(itemName);
474         } catch (ItemNotFoundException e1) {
475             return null;
476         }
477     }
478
479     @Override
480     public List<PersistenceStrategy> getDefaultStrategies() {
481         return List.of(PersistenceStrategy.Globals.RESTORE, PersistenceStrategy.Globals.CHANGE);
482     }
483
484     @Override
485     public void store(Item item) {
486         store(item, null);
487     }
488
489     @Override
490     public void store(Item item, @Nullable String alias) {
491         // Timestamp and capture state immediately as rest of the store is asynchronous (state might change in between)
492         ZonedDateTime time = ZonedDateTime.now();
493
494         logIfManyQueuedTasks();
495         if (!(item instanceof GenericItem)) {
496             return;
497         }
498         if (item.getState() instanceof UnDefType) {
499             logger.debug("Undefined item state received. Not storing item {}.", item.getName());
500             return;
501         }
502         if (!isReadyToStore()) {
503             logger.warn("Not ready to store (config error?), not storing item {}.", item.getName());
504             return;
505         }
506         // Get Item describing the real type of data
507         // With non-group items this is same as the argument item. With Group items, this is item describing the type of
508         // state stored in the group.
509         final Item itemTemplate;
510         try {
511             itemTemplate = getEffectiveItem(item);
512         } catch (IllegalStateException e) {
513             // Exception is raised when underlying item type cannot be determined with Group item
514             // Logged already
515             return;
516         }
517
518         String effectiveName = (alias != null) ? alias : item.getName();
519
520         // We do not want to rely item.state since async context below can execute much later.
521         // We 'copy' the item for local use. copyItem also normalizes the unit with NumberItems.
522         final GenericItem copiedItem = copyItem(itemTemplate, item, effectiveName, null);
523
524         resolveTableSchema().thenAcceptAsync(resolved -> {
525             if (!resolved) {
526                 logger.warn("Table schema not resolved, not storing item {}.", copiedItem.getName());
527                 return;
528             }
529
530             DynamoDbEnhancedAsyncClient localClient = client;
531             DynamoDbAsyncClient localLowlevelClient = lowLevelClient;
532             DynamoDBConfig localConfig = dbConfig;
533             DynamoDBTableNameResolver localTableNameResolver = tableNameResolver;
534             if (!isProperlyConfigured || localClient == null || localLowlevelClient == null || localConfig == null
535                     || localTableNameResolver == null) {
536                 logger.warn("Not ready to store (config error?), not storing item {}.", item.getName());
537                 return;
538             }
539
540             Integer expireDays = localConfig.getExpireDays();
541
542             final DynamoDBItem<?> dto;
543             switch (localTableNameResolver.getTableSchema()) {
544                 case NEW:
545                     dto = AbstractDynamoDBItem.fromStateNew(copiedItem, time, expireDays);
546                     break;
547                 case LEGACY:
548                     dto = AbstractDynamoDBItem.fromStateLegacy(copiedItem, time);
549                     break;
550                 default:
551                     throw new IllegalStateException("Unexpected. Bug");
552             }
553             logger.trace("store() called with item {} {} '{}', which was converted to DTO {}",
554                     copiedItem.getClass().getSimpleName(), effectiveName, copiedItem.getState(), dto);
555             dto.accept(new DynamoDBItemVisitor<TableCreatingPutItem<? extends DynamoDBItem<?>>>() {
556
557                 @Override
558                 public TableCreatingPutItem<? extends DynamoDBItem<?>> visit(
559                         DynamoDBBigDecimalItem dynamoBigDecimalItem) {
560                     return new TableCreatingPutItem<DynamoDBBigDecimalItem>(DynamoDBPersistenceService.this,
561                             dynamoBigDecimalItem, getTable(DynamoDBBigDecimalItem.class));
562                 }
563
564                 @Override
565                 public TableCreatingPutItem<? extends DynamoDBItem<?>> visit(DynamoDBStringItem dynamoStringItem) {
566                     return new TableCreatingPutItem<DynamoDBStringItem>(DynamoDBPersistenceService.this,
567                             dynamoStringItem, getTable(DynamoDBStringItem.class));
568                 }
569             }).putItemAsync();
570         }, executor).exceptionally(e -> {
571             logger.error("Unexcepted error", e);
572             return null;
573         });
574     }
575
576     private Item getEffectiveItem(Item item) {
577         final Item effectiveItem;
578         if (item instanceof GroupItem) {
579             Item baseItem = ((GroupItem) item).getBaseItem();
580             if (baseItem == null) {
581                 // if GroupItem:<ItemType> is not defined in
582                 // *.items using StringType
583                 logger.debug(
584                         "Cannot detect ItemType for {} because the GroupItems' base type isn't set in *.items File.",
585                         item.getName());
586                 Iterator<Item> firstGroupMemberItem = ((GroupItem) item).getMembers().iterator();
587                 if (firstGroupMemberItem.hasNext()) {
588                     effectiveItem = firstGroupMemberItem.next();
589                 } else {
590                     throw new IllegalStateException("GroupItem " + item.getName()
591                             + " does not have children nor base item set, cannot determine underlying item type. Aborting!");
592                 }
593             } else {
594                 effectiveItem = baseItem;
595             }
596         } else {
597             effectiveItem = item;
598         }
599         return effectiveItem;
600     }
601
602     /**
603      * Copy item and optionally override name and state
604      *
605      * State is normalized to source item's unit with Quantity NumberItems and QuantityTypes
606      *
607      * @param itemTemplate 'template item' to be used to construct the new copy. It is also used to determine UoM unit
608      *            and get GenericItem.type
609      * @param item item that is used to acquire name and state
610      * @param nameOverride name override for the resulting copy
611      * @param stateOverride state override for the resulting copy
612      * @throws IllegalArgumentException when state is QuantityType and not compatible with item
613      */
614     static GenericItem copyItem(Item itemTemplate, Item item, @Nullable String nameOverride,
615             @Nullable State stateOverride) {
616         final GenericItem copiedItem;
617         try {
618             if (itemTemplate instanceof NumberItem) {
619                 copiedItem = (GenericItem) itemTemplate.getClass().getDeclaredConstructor(String.class, String.class)
620                         .newInstance(itemTemplate.getType(), nameOverride == null ? item.getName() : nameOverride);
621             } else {
622                 copiedItem = (GenericItem) itemTemplate.getClass().getDeclaredConstructor(String.class)
623                         .newInstance(nameOverride == null ? item.getName() : nameOverride);
624             }
625
626         } catch (InstantiationException | IllegalAccessException | IllegalArgumentException | InvocationTargetException
627                 | NoSuchMethodException | SecurityException e) {
628             throw new IllegalArgumentException(item.toString(), e);
629         }
630         State state = stateOverride == null ? item.getState() : stateOverride;
631         if (state instanceof QuantityType<?> && itemTemplate instanceof NumberItem) {
632             Unit<?> itemUnit = ((NumberItem) itemTemplate).getUnit();
633             if (itemUnit != null) {
634                 State convertedState = ((QuantityType<?>) state).toUnit(itemUnit);
635                 if (convertedState == null) {
636                     logger.error("Unexpected unit conversion failure: {} to item unit {}", state, itemUnit);
637                     throw new IllegalArgumentException(
638                             String.format("Unexpected unit conversion failure: %s to item unit %s", state, itemUnit));
639                 }
640                 state = convertedState;
641             }
642         }
643         copiedItem.setState(state);
644         return copiedItem;
645     }
646
647     private void logIfManyQueuedTasks() {
648         if (executor instanceof ThreadPoolExecutor) {
649             ThreadPoolExecutor localExecutor = (ThreadPoolExecutor) executor;
650             if (localExecutor.getQueue().size() >= 5) {
651                 logger.trace("executor queue size: {}, remaining space {}. Active threads {}",
652                         localExecutor.getQueue().size(), localExecutor.getQueue().remainingCapacity(),
653                         localExecutor.getActiveCount());
654             } else if (localExecutor.getQueue().size() >= 50) {
655                 logger.warn(
656                         "Many ({}) tasks queued in executor! This might be sign of bad design or bug in the addon code.",
657                         localExecutor.getQueue().size());
658             }
659         }
660     }
661
662     private String filterToString(FilterCriteria filter) {
663         return String.format(
664                 "FilterCriteria@%s(item=%s, pageNumber=%d, pageSize=%d, time=[%s, %s, %s], state=[%s, %s of %s] )",
665                 System.identityHashCode(filter), filter.getItemName(), filter.getPageNumber(), filter.getPageSize(),
666                 filter.getBeginDate(), filter.getEndDate(), filter.getOrdering(), filter.getOperator(),
667                 filter.getState(), filter.getState() == null ? "null" : filter.getState().getClass().getSimpleName());
668     }
669 }