| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| package org.apache.samza.storage.kv.inmemory |
| |
| import com.google.common.primitives.UnsignedBytes |
| import org.apache.samza.util.Logging |
| import org.apache.samza.storage.kv.{KeyValueStoreMetrics, KeyValueIterator, Entry, KeyValueStore} |
| import java.util |
| |
| /** |
| * In memory implementation of a key value store. |
| * |
| * This uses a TreeMap to store the keys in order |
| * |
| * @param metrics A metrics instance to publish key-value store related statistics |
| */ |
| class InMemoryKeyValueStore(val metrics: KeyValueStoreMetrics = new KeyValueStoreMetrics) |
| extends KeyValueStore[Array[Byte], Array[Byte]] with Logging { |
| |
| val underlying = new util.TreeMap[Array[Byte], Array[Byte]] (UnsignedBytes.lexicographicalComparator()) |
| |
| override def flush(): Unit = { |
| // No-op for In memory store. |
| metrics.flushes.inc |
| } |
| |
| override def close(): Unit = Unit |
| |
| private class InMemoryIterator (val iter: util.Iterator[util.Map.Entry[Array[Byte], Array[Byte]]]) |
| extends KeyValueIterator[Array[Byte], Array[Byte]] { |
| |
| override def close(): Unit = Unit |
| |
| override def remove(): Unit = iter.remove() |
| |
| override def next(): Entry[Array[Byte], Array[Byte]] = { |
| val n = iter.next() |
| if (n != null && n.getKey != null) { |
| metrics.bytesRead.inc(n.getKey.size) |
| } |
| if (n != null && n.getValue != null) { |
| metrics.bytesRead.inc(n.getValue.size) |
| } |
| new Entry(n.getKey, n.getValue) |
| } |
| |
| override def hasNext: Boolean = iter.hasNext |
| } |
| |
| override def all(): KeyValueIterator[Array[Byte], Array[Byte]] = { |
| metrics.alls.inc |
| |
| new InMemoryIterator(underlying.entrySet().iterator()) |
| } |
| |
| override def range(from: Array[Byte], to: Array[Byte]): KeyValueIterator[Array[Byte], Array[Byte]] = { |
| metrics.ranges.inc |
| require(from != null && to != null, "Null bound not allowed.") |
| |
| new InMemoryIterator(underlying.subMap(from, to).entrySet().iterator()) |
| } |
| |
| override def delete(key: Array[Byte]): Unit = { |
| metrics.deletes.inc |
| put(key, null) |
| } |
| |
| override def deleteAll(keys: java.util.List[Array[Byte]]) = { |
| KeyValueStore.Extension.deleteAll(this, keys) |
| } |
| |
| override def putAll(entries: util.List[Entry[Array[Byte], Array[Byte]]]): Unit = { |
| // TreeMap's putAll requires a map, so we'd need to iterate over all the entries anyway |
| // to use it, in order to putAll here. Therefore, just iterate here. |
| val iter = entries.iterator() |
| while(iter.hasNext) { |
| val next = iter.next() |
| put(next.getKey, next.getValue) |
| } |
| } |
| |
| override def put(key: Array[Byte], value: Array[Byte]): Unit = { |
| metrics.puts.inc |
| require(key != null, "Null key not allowed.") |
| if (value == null) { |
| metrics.deletes.inc |
| underlying.remove(key) |
| } else { |
| metrics.bytesWritten.inc(key.size + value.size) |
| underlying.put(key, value) |
| } |
| } |
| |
| override def get(key: Array[Byte]): Array[Byte] = { |
| metrics.gets.inc |
| require(key != null, "Null key not allowed.") |
| val found = underlying.get(key) |
| if (found != null) { |
| metrics.bytesRead.inc(found.size) |
| } |
| found |
| } |
| |
| override def getAll(keys: java.util.List[Array[Byte]]): java.util.Map[Array[Byte], Array[Byte]] = { |
| KeyValueStore.Extension.getAll(this, keys); |
| } |
| } |