| /* |
| * 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.ignite.scalar |
| |
| import org.apache.ignite._ |
| import org.apache.ignite.cache.CacheMode |
| import org.apache.ignite.cache.query.annotations.{QuerySqlField, QueryTextField} |
| import org.apache.ignite.cluster.ClusterNode |
| import org.apache.ignite.configuration.{CacheConfiguration, IgniteConfiguration} |
| import org.apache.ignite.internal.IgniteVersionUtils._ |
| import org.jetbrains.annotations.Nullable |
| |
| import java.net.URL |
| import java.util.UUID |
| |
| import scala.annotation.meta.field |
| |
| /** |
| * {{{ |
| * ________ ______ ______ _______ |
| * __ ___/_____________ ____ /______ _________ __/__ \ __ __ \ |
| * _____ \ _ ___/_ __ `/__ / _ __ `/__ ___/ ____/ / _ / / / |
| * ____/ / / /__ / /_/ / _ / / /_/ / _ / _ __/___/ /_/ / |
| * /____/ \___/ \__,_/ /_/ \__,_/ /_/ /____/_(_)____/ |
| * |
| * }}} |
| * |
| * ==Overview== |
| * `scalar` is the main object that encapsulates Scalar DSL. It includes global functions |
| * on "scalar" keyword, helper converters as well as necessary implicit conversions. `scalar` also |
| * mimics many methods in `Ignite` class from Java side. |
| * |
| * The idea behind Scalar DSL - '''zero additional logic and only conversions''' implemented |
| * using Scala "Pimp" pattern. Note that most of the Scalar DSL development happened on Java |
| * side of Ignite 3.0 product line - Java APIs had to be adjusted quite significantly to |
| * support natural adaptation of functional APIs. That basically means that all functional |
| * logic must be available on Java side and Scalar only provides conversions from Scala |
| * language constructs to Java constructs. Note that currently Ignite supports Scala 2.8 |
| * and up only. |
| * |
| * This design approach ensures that Java side does not starve and usage paradigm |
| * is mostly the same between Java and Scala - yet with full power of Scala behind. |
| * In other words, Scalar only adds Scala specifics, but not greatly altering semantics |
| * of how Ignite APIs work. Most of the time the code in Scalar can be written in |
| * Java in almost the same number of lines. |
| * |
| * ==Suffix '$' In Names== |
| * Symbol `$` is used in names when they conflict with the names in the base Java class |
| * that Scala pimp is shadowing or with Java package name that your Scala code is importing. |
| * Instead of giving two different names to the same function we've decided to simply mark |
| * Scala's side method with `$` suffix. |
| * |
| * ==Importing== |
| * Scalar needs to be imported in a proper way so that necessary objects and implicit |
| * conversions got available in the scope: |
| * <pre name="code" class="scala"> |
| * import org.apache.ignite.scalar._ |
| * import scalar._ |
| * </pre> |
| * This way you import object `scalar` as well as all methods declared or inherited in that |
| * object as well. |
| * |
| * ==Examples== |
| * Here are few short examples of how Scalar can be used to program routine distributed |
| * task. All examples below use default Ignite configuration and default grid. All these |
| * examples take an implicit advantage of auto-discovery and failover, load balancing and |
| * collision resolution, zero deployment and many other underlying technologies in the |
| * Ignite - while remaining absolutely distilled to the core domain logic. |
| * |
| * This code snippet prints out full topology: |
| * <pre name="code" class="scala"> |
| * scalar { |
| * grid$ foreach (n => println("Node: " + n.id8)) |
| * } |
| * </pre> |
| * The obligatory example - cloud enabled `Hello World!`. It splits the phrase |
| * into multiple words and prints each word on a separate grid node: |
| * <pre name="code" class="scala"> |
| * scalar { |
| * grid$ *< (SPREAD, (for (w <- "Hello World!".split(" ")) yield () => println(w))) |
| * } |
| * </pre> |
| * This example broadcasts message to all nodes: |
| * <pre name="code" class="scala"> |
| * scalar { |
| * grid$ *< (BROADCAST, () => println("Broadcasting!!!")) |
| * } |
| * </pre> |
| * This example "greets" remote nodes only (note usage of Java-side closure): |
| * <pre name="code" class="scala"> |
| * scalar { |
| * val me = grid$.localNode.id |
| * grid$.remoteProjection() *< (BROADCAST, F.println("Greetings from: " + me)) |
| * } |
| * </pre> |
| * |
| * Next example creates a function that calculates lengths of the string |
| * using MapReduce type of processing by splitting the input string into |
| * multiple substrings, calculating each substring length on the remote |
| * node and aggregating results for the final length of the original string: |
| * <pre name="code" class="scala"> |
| * def count(msg: String) = |
| * grid$ @< (SPREAD, for (w <- msg.split(" ")) yield () => w.length, (s: Seq[Int]) => s.sum) |
| * </pre> |
| * This example shows a simple example of how Scalar can be used to work with in-memory data grid: |
| * <pre name="code" class="scala"> |
| * scalar { |
| * val t = cache$[Symbol, Double]("partitioned") |
| * t += ('symbol -> 2.0) |
| * t -= ('symbol) |
| * } |
| * </pre> |
| */ |
| object scalar extends ScalarConversions { |
| /** Type alias for `QuerySqlField`. */ |
| type ScalarCacheQuerySqlField = QuerySqlField @field |
| |
| /** Type alias for `QueryTextField`. */ |
| type ScalarCacheQueryTextField = QueryTextField @field |
| |
| /** |
| * Prints Scalar ASCII-logo. |
| */ |
| def logo() { |
| val NL = System getProperty "line.separator" |
| |
| val s = |
| " ________ ______ " + NL + |
| " __ ___/_____________ ____ /______ _________ " + NL + |
| " _____ \\ _ ___/_ __ `/__ / _ __ `/__ ___/ " + NL + |
| " ____/ / / /__ / /_/ / _ / / /_/ / _ / " + NL + |
| " /____/ \\___/ \\__,_/ /_/ \\__,_/ /_/ " + NL + NL + |
| " IGNITE SCALAR" + |
| " " + COPYRIGHT + NL |
| |
| println(s) |
| } |
| |
| /** |
| * Note that grid instance will be stopped with cancel flat set to `true`. |
| * |
| * @param g Grid instance. |
| * @param body Closure with grid instance as body's parameter. |
| */ |
| private def init[T](g: Ignite, body: Ignite => T): T = { |
| assert(g != null, body != null) |
| |
| try { |
| body(g) |
| } |
| finally { |
| Ignition.stop(g.name, true) |
| } |
| } |
| |
| /** |
| * Note that grid instance will be stopped with cancel flat set to `true`. |
| * |
| * @param g Grid instance. |
| * @param body Passed by name body. |
| */ |
| private def init0[T](g: Ignite, body: => T): T = { |
| assert(g != null) |
| |
| try { |
| body |
| } |
| finally { |
| Ignition.stop(g.name, true) |
| } |
| } |
| |
| /** |
| * Executes given closure within automatically managed default grid instance. |
| * If default grid is already started the passed in closure will simply |
| * execute. |
| * |
| * @param body Closure to execute within automatically managed default grid instance. |
| */ |
| def apply(body: Ignite => Unit) { |
| if (!isStarted) init(Ignition.start, body) else body(ignite$) |
| } |
| |
| /** |
| * Executes given closure within automatically managed default grid instance. |
| * If default grid is already started the passed in closure will simply |
| * execute. |
| * |
| * @param body Closure to execute within automatically managed default grid instance. |
| */ |
| def apply[T](body: Ignite => T): T = |
| if (!isStarted) init(Ignition.start, body) else body(ignite$) |
| |
| /** |
| * Executes given closure within automatically managed default grid instance. |
| * If default grid is already started the passed in closure will simply |
| * execute. |
| * |
| * @param body Closure to execute within automatically managed default grid instance. |
| */ |
| def apply[T](body: => T): T = |
| if (!isStarted) init0(Ignition.start, body) else body |
| |
| /** |
| * Executes given closure within automatically managed default grid instance. |
| * If default grid is already started the passed in closure will simply |
| * execute. |
| * |
| * @param body Closure to execute within automatically managed grid instance. |
| */ |
| def apply(body: => Unit) { |
| if (!isStarted) init0(Ignition.start, body) else body |
| } |
| |
| /** |
| * Executes given closure within automatically managed grid instance. |
| * |
| * @param springCfgPath Spring XML configuration file path or URL. |
| * @param body Closure to execute within automatically managed grid instance. |
| */ |
| def apply(springCfgPath: String)(body: => Unit) { |
| init0(Ignition.start(springCfgPath), body) |
| } |
| |
| /** |
| * Executes given closure within automatically managed grid instance. |
| * |
| * @param cfg Grid configuration instance. |
| * @param body Closure to execute within automatically managed grid instance. |
| */ |
| def apply(cfg: IgniteConfiguration)(body: => Unit) { |
| init0(Ignition.start(cfg), body) |
| } |
| |
| /** |
| * Executes given closure within automatically managed grid instance. |
| * |
| * @param springCfgUrl Spring XML configuration file URL. |
| * @param body Closure to execute within automatically managed grid instance. |
| */ |
| def apply(springCfgUrl: URL)(body: => Unit) { |
| init0(Ignition.start(springCfgUrl), body) |
| } |
| |
| /** |
| * Gets named cache from default grid. |
| * |
| * @param cacheName Name of the cache to get. |
| */ |
| @inline def cache$[K, V](cacheName: String): Option[IgniteCache[K, V]] = |
| Option(Ignition.ignite.cache(cacheName)) |
| |
| /** |
| * Creates cache with specified parameters in default grid. |
| * |
| * @param cacheName Name of the cache to get. |
| */ |
| @inline def createCache$[K, V](cacheName: String, cacheMode: CacheMode = CacheMode.PARTITIONED, |
| indexedTypes: Seq[Class[_]] = Seq.empty): IgniteCache[K, V] = { |
| val cfg = new CacheConfiguration[K, V]() |
| |
| cfg.setName(cacheName) |
| cfg.setCacheMode(cacheMode) |
| cfg.setIndexedTypes(indexedTypes:_*) |
| |
| Ignition.ignite.createCache(cfg) |
| } |
| |
| /** |
| * Destroy cache with specified name. |
| * |
| * @param cacheName Name of the cache to destroy. |
| */ |
| @inline def destroyCache$(cacheName: String) = { |
| Ignition.ignite.destroyCache(cacheName) |
| } |
| |
| /** |
| * Gets named cache from specified grid. |
| * |
| * @param igniteInstanceName Name of the Ignite instance. |
| * @param cacheName Name of the cache to get. |
| */ |
| @inline def cache$[K, V](@Nullable igniteInstanceName: String, |
| cacheName: String): Option[IgniteCache[K, V]] = |
| ignite$(igniteInstanceName) match { |
| case Some(g) => Option(g.cache(cacheName)) |
| case None => None |
| } |
| |
| /** |
| * Gets a new instance of data streamer associated with given cache name. |
| * |
| * @param cacheName Cache name (`null` for default cache). |
| * @param bufSize Per node buffer size. |
| * @return New instance of data streamer. |
| */ |
| @inline def dataStreamer$[K, V]( |
| cacheName: String, |
| bufSize: Int): IgniteDataStreamer[K, V] = { |
| val dl = ignite$.dataStreamer[K, V](cacheName) |
| |
| dl.perNodeBufferSize(bufSize) |
| |
| dl |
| } |
| |
| /** |
| * Gets default grid instance. |
| */ |
| @inline def ignite$: Ignite = Ignition.ignite |
| |
| /** |
| * Gets node ID as ID8 string. |
| */ |
| def nid8$(node: ClusterNode) = node.id().toString.take(8).toUpperCase |
| |
| /** |
| * Gets named Ignite instance. |
| * |
| * @param name Ignite instance name. |
| */ |
| @inline def ignite$(@Nullable name: String): Option[Ignite] = |
| try { |
| Option(Ignition.ignite(name)) |
| } |
| catch { |
| case _: IllegalStateException => None |
| } |
| |
| /** |
| * Gets grid for given node ID. |
| * |
| * @param locNodeId Local node ID for which to get grid instance option. |
| */ |
| @inline def grid$(locNodeId: UUID): Option[Ignite] = { |
| assert(locNodeId != null) |
| |
| try { |
| Option(Ignition.ignite(locNodeId)) |
| } |
| catch { |
| case _: IllegalStateException => None |
| } |
| } |
| |
| /** |
| * Tests if specified grid is started. |
| * |
| * @param name Gird name. |
| */ |
| def isStarted(@Nullable name: String) = |
| Ignition.state(name) == IgniteState.STARTED |
| |
| /** |
| * Tests if specified grid is stopped. |
| * |
| * @param name Gird name. |
| */ |
| def isStopped(@Nullable name: String) = |
| Ignition.state(name) == IgniteState.STOPPED |
| |
| /** |
| * Tests if default grid is started. |
| */ |
| def isStarted = |
| Ignition.state == IgniteState.STARTED |
| |
| /** |
| * Tests if default grid is stopped. |
| */ |
| def isStopped = |
| Ignition.state == IgniteState.STOPPED |
| |
| /** |
| * Stops given Ignite instance and specified cancel flag. |
| * If specified Ignite instance is already stopped - it's no-op. |
| * |
| * @param name Ignite instance name to cancel. |
| * @param cancel Whether or not to cancel all currently running jobs. |
| */ |
| def stop(@Nullable name: String, cancel: Boolean) = |
| if (isStarted(name)) |
| Ignition.stop(name, cancel) |
| |
| /** |
| * Stops default grid with given cancel flag. |
| * If default grid is already stopped - it's no-op. |
| * |
| * @param cancel Whether or not to cancel all currently running jobs. |
| */ |
| def stop(cancel: Boolean) = |
| if (isStarted) Ignition.stop(cancel) |
| |
| /** |
| * Stops default grid with cancel flag set to `true`. |
| * If default grid is already stopped - it's no-op. |
| */ |
| def stop() = |
| if (isStarted) Ignition.stop(true) |
| |
| /** |
| * Sets daemon flag to grid factory. Note that this method should be called |
| * before grid instance starts. |
| * |
| * @param f Daemon flag to set. |
| */ |
| def daemon(f: Boolean) { |
| Ignition.setDaemon(f) |
| } |
| |
| /** |
| * Gets daemon flag set in the grid factory. |
| */ |
| def isDaemon = |
| Ignition.isDaemon |
| |
| /** |
| * Starts default grid. It's no-op if default grid is already started. |
| * |
| * @return Started grid. |
| */ |
| def start(): Ignite = { |
| if (!isStarted) Ignition.start else ignite$ |
| } |
| |
| /** |
| * Starts grid with given parameter(s). |
| * |
| * @param springCfgPath Spring XML configuration file path or URL. |
| * @return Started grid. If Spring configuration contains multiple grid instances, |
| * then the 1st found instance is returned. |
| */ |
| def start(@Nullable springCfgPath: String): Ignite = { |
| Ignition.start(springCfgPath) |
| } |
| |
| /** |
| * Starts grid with given parameter(s). |
| * |
| * @param cfg Grid configuration. This cannot be `null`. |
| * @return Started grid. |
| */ |
| def start(cfg: IgniteConfiguration): Ignite = { |
| Ignition.start(cfg) |
| } |
| |
| /** |
| * Starts grid with given parameter(s). |
| * |
| * @param springCfgUrl Spring XML configuration file URL. |
| * @return Started grid. |
| */ |
| def start(springCfgUrl: URL): Ignite = { |
| Ignition.start(springCfgUrl) |
| } |
| } |
