diff --git a/examples/example-actors/.idea/workspace.xml b/examples/example-actors/.idea/workspace.xml
index 63cad3dacd2..47070f8b8ef 100644
--- a/examples/example-actors/.idea/workspace.xml
+++ b/examples/example-actors/.idea/workspace.xml
@@ -76,16 +76,7 @@
-
-
-
-
-
-
-
-
-
-
+
@@ -179,8 +170,8 @@
-
+
@@ -216,26 +207,6 @@
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
@@ -348,20 +319,40 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-
+
-
+
+
-
-
+
+
@@ -429,24 +420,6 @@
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
@@ -479,6 +452,24 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
@@ -560,7 +551,7 @@
-
+
@@ -576,7 +567,7 @@
-
+
@@ -673,6 +664,20 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
@@ -703,30 +708,9 @@
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
diff --git a/examples/example-actors/src/org/jetbrains/examples/actors/Actors.kt b/examples/example-actors/src/org/jetbrains/examples/actors/Actors.kt
index a116da31848..a643254ba89 100644
--- a/examples/example-actors/src/org/jetbrains/examples/actors/Actors.kt
+++ b/examples/example-actors/src/org/jetbrains/examples/actors/Actors.kt
@@ -7,6 +7,127 @@ import java.util.concurrent.*
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater
import java.util.Date
+/*
+Message handling class which process only one message at any given moment
+
+Three main ways to use it
+- post and forget
+- post and await result
+- post and assign callback to be called when message processed
+*/
+abstract class Actor(protected val executor: Executor) {
+ // we can not do it private or protected
+ // otherwise updater defined in class object will not be able to access it
+ volatile var queue : FunctionalQueue = emptyQueue
+
+ /*
+ Handle message and return result
+ This method guaranteed to be executed only one per object at any given time
+ */
+ protected abstract fun onMessage(message: Any) : Any?
+
+ /*
+ Post message to the actor.
+ The method returns immediately and the message itself will be processed later
+ */
+ fun post(message: Any) {
+ while(true) {
+ val q = queue
+ if(q.empty) {
+ if(queueUpdater.compareAndSet(this, q, busyEmptyQueue)) {
+ executor.execute { process(message) }
+ break
+ }
+ }
+ else {
+ val newQueue = (if(q == busyEmptyQueue) emptyQueue else q) add message
+ if(queueUpdater.compareAndSet(this, q, newQueue)) {
+ break
+ }
+ }
+ }
+ }
+
+ /*
+ Post message to the actor and schedule callback to be executed on given executor when message processed
+ */
+ fun post(message: Any, executor: Executor = this.executor, callback: (Any?)->Unit) =
+ post(Callback(message, executor, callback))
+
+ /*
+ Send message to the actor and await result
+ */
+ fun send(message: Any) : Any? {
+ val request = Request(message)
+ post(request)
+ request.await()
+ return request.result
+ }
+
+ private fun nextMessage() {
+ while(true) {
+ val q = queue
+ if(q == busyEmptyQueue) {
+ if(queueUpdater.compareAndSet(this, busyEmptyQueue, emptyQueue)) {
+ break;
+ }
+ }
+ else {
+ val removed = q.removeFirst()
+ val newQueue = if(removed._2.empty) busyEmptyQueue else removed._2
+ if(queueUpdater.compareAndSet(this, q, newQueue)) {
+ executor.execute { process(removed._1) }
+ break;
+ }
+ }
+ }
+ }
+
+ private fun process(message: Any) {
+ when(message) {
+ is Request -> {
+ message.result = onMessage(message.message)
+ message.countDown()
+ }
+ is Callback -> {
+ val result = onMessage(message.message)
+ message.executor execute {
+ val callback = message.callback;
+ callback(result)
+ }
+ }
+ else -> onMessage(message)
+ }
+
+ nextMessage()
+ }
+
+ /*
+ Create actor on the same executor
+ */
+ fun actor(handler: (Any)->Any?) = executor.actor(handler)
+
+ class object {
+ val queueUpdater = AtomicReferenceFieldUpdater.newUpdater(typeinfo.javaClassForType,typeinfo>().javaClassForType, "queue").sure()
+ val emptyQueue = FunctionalQueue()
+ val busyEmptyQueue = FunctionalQueue() add "busy empty queue"
+
+ class Request(val message: Any) : java.util.concurrent.CountDownLatch(1) {
+ var result: Any? = null
+ }
+
+ class Callback(val message: Any, val executor: Executor, val callback: (Any?) -> Unit)
+ }
+}
+
+fun Executor.actor(handler: (Any)->Any?) : Actor = object: Actor(this) {
+ override fun onMessage(message: Any) {
+ handler(message)
+ }
+}
+
+fun singleThreadActor(handler: (Any)->Any?) : Actor = Executors.newSingleThreadExecutor().sure().actor(handler)
+
class App() : Actor(Executors.newFixedThreadPool(10).sure()) {
private val logger = executor.actor { message ->
println("${Date()}:\t\t$message")
diff --git a/stdlib/ktSrc/concurrent/Actors.kt b/stdlib/ktSrc/concurrent/Actors.kt
deleted file mode 100644
index cae87659523..00000000000
--- a/stdlib/ktSrc/concurrent/Actors.kt
+++ /dev/null
@@ -1,130 +0,0 @@
-package std.concurrent
-
-import std.concurrent.*
-import std.util.*
-
-import java.util.concurrent.*
-import java.util.concurrent.atomic.AtomicReferenceFieldUpdater
-import java.util.Date
-
-/*
-Message handling class which process only one message at any given moment
-
-Three main ways to use it
-- post and forget
-- post and await result
-- post and assign callback when message processed
-*/
-abstract class Actor(protected val executor: Executor) {
- // we can not do it private or protected
- // otherwise updater defined in class object will not be able to access it
- volatile var queue : FunctionalQueue = emptyQueue
-
- /*
- Handle message and return result
- This method guaranteed to be executed only one per object at any given time
- */
- protected abstract fun onMessage(message: Any) : Any?
-
- /*
- Post message to the actor.
- The method returns immediately and the message itself will be processed later
- */
- fun post(message: Any) {
- while(true) {
- val q = queue
- if(q.empty) {
- if(queueUpdater.compareAndSet(this, q, busyEmptyQueue)) {
- executor.execute { process(message) }
- break
- }
- }
- else {
- val newQueue = (if(q == busyEmptyQueue) emptyQueue else q) add message
- if(queueUpdater.compareAndSet(this, q, newQueue)) {
- break
- }
- }
- }
- }
-
- /*
- Post message to the actor and schedule callback to be executed on given executor when message processed
- */
- fun post(message: Any, executor: Executor = this.executor, callback: (Any?)->Unit) =
- post(Callback(message, executor, callback))
-
- /*
- Send message to the actor and await result
- */
- fun send(message: Any) : Any? {
- val request = Request(message)
- post(request)
- request.await()
- return request.result
- }
-
- private fun nextMessage() {
- while(true) {
- val q = queue
- if(q == busyEmptyQueue) {
- if(queueUpdater.compareAndSet(this, busyEmptyQueue, emptyQueue)) {
- break;
- }
- }
- else {
- val removed = q.removeFirst()
- val newQueue = if(removed._2.empty) busyEmptyQueue else removed._2
- if(queueUpdater.compareAndSet(this, q, newQueue)) {
- executor.execute { process(removed._1) }
- break;
- }
- }
- }
- }
-
- private fun process(message: Any) {
- when(message) {
- is Request -> {
- message.result = onMessage(message.message)
- message.countDown()
- }
- is Callback -> {
- val result = onMessage(message.message)
- message.executor execute {
- val callback = message.callback;
- callback(result)
- }
- }
- else -> onMessage(message)
- }
-
- nextMessage()
- }
-
- /*
- Create actor on the same executor
- */
- fun actor(handler: (Any)->Any?) = executor.actor(handler)
-
- class object {
- val queueUpdater = AtomicReferenceFieldUpdater.newUpdater(typeinfo.javaClassForType,typeinfo>().javaClassForType, "queue").sure()
- val emptyQueue = FunctionalQueue()
- val busyEmptyQueue = FunctionalQueue() add "busy empty queue"
-
- class Request(val message: Any) : java.util.concurrent.CountDownLatch(1) {
- var result: Any? = null
- }
-
- class Callback(val message: Any, val executor: Executor, val callback: (Any?) -> Unit)
- }
-}
-
-fun Executor.actor(handler: (Any)->Any?) : Actor = object: Actor(this) {
- override fun onMessage(message: Any) {
- handler(message)
- }
-}
-
-fun singleThreadActor(handler: (Any)->Any?) : Actor = Executors.newSingleThreadExecutor().sure().actor(handler)
-