Actors.ns

Newspeak3
'Actors'
class Actors usingPlatform: p vm: vm = (|
private SharedQueue2 = p squeak SharedQueue2.
private WeakMap = p squeak WeakKeyDictionary.
private List = p collections List.
private Transcript = p squeak Transcript.
private SqueakDelay = p squeak Delay.

private Processor <ProcessScheduler> = p squeak Processor.
private UnhandledError <ProcessScheduler> = p squeak UnhandledError.
private primitives <VMMirror> = vm.

private brazil = p brazil.
public (* bogus - temporary access to a default actor *) uiActor = UIActor new.

private internalRefs <WeakMap[FarReference, InternalFarReference]> = WeakMap new.

public Promise = PromiseUtils new.
|p squeak Project uiProcess actor: uiActor.
p actors: self) (
public class Actor named: debugName <String> = (|
private (* N.B. Do not leak to subinstances. *) internalActor = (ProcessActor named: debugName) startEventLoop.
|) (
public interrupt = (
	unimplemented
)
public onError: onError <[:E]> = (
	unimplemented.
)
public seed: value <Value> ^<FarReference[Value]> = (
	(* Creating a far reference to a PassByCopy is questionable. It may be that this should be seed: receiver perform: selector with: argument. *)
	(isValueObject: value) ifFalse: [Error signal: 'Cannot seed an actor with a non-Value.'].
	^createFarReferenceTo: value in: internalActor for: currentActor.
)
public terminate = (
	unimplemented
)
) : (
)
protected class EventualSendReactor selector: s arguments: a resolver: r = (
(* An eventual-send to a promise.

Note that all slots contain objects that belong to the actor which performed the eventual send. *)
|
protected selector <Symbol> = s.
protected arguments <Array> = a.
protected resolver <InternalResolver> = r.
|) (
public reactBroken: error = (
	(* Broken promise contagion. *)
	resolver __owner == currentActor ifFalse: [halt].
	(* Should this be immediate? *)
	(* resolver internalBreak: error. ? *)
	currentActor enqueueMessage:
		(PendingDelivery
			receiver: resolver
			selector: #internalResolve:
			arguments: {InternalBrokenRef problem: error __owner: resolver __owner}
			resolver: nil)
)
public reactMoreResolved: internalRef = (
	resolver internalResolve: (asInternalRef: (internalRef eventualSend: selector arguments: arguments)).
)
public reactResolved: value = (
	resolver __owner == currentActor ifFalse: [halt].
	resolver internalResolve:
		(asInternalRef: (currentActor
			eventualSendTo: value
			selector: selector
			arguments: arguments))
)
) : (
)
class InternalActor = () (
public currentActor = (
	(* BOGUS. Polymorphic with other actor system. *)
	^self
)
public eventualSendTo: receiver selector: selector arguments: arguments ^<Ref> = (
	(* Sent from the compilation of an eventual send, via the same method in ProcessScheduler. *)
	^(asInternalRef: receiver) eventualSend: selector arguments: arguments
)
) : (
)
class InternalBrokenRef problem: p __owner: o = InternalRef __owner: o (|
protected problem = p.
|) (
public addReactor: reactor = (
	reactor reactBroken: problem.
)
public checkCycle ^<Boolean> = (
	^false
)
public eventualSend: selector <Symbol> arguments: arguments <Array> ^<Ref> = (
	^self externalRef (* I.e., broken with same problem. Should this wait for the next turn? *)
)
public immediateSend: message <Message> ^<Object> = (
	^Error signal: 'Cannot immediate-send to a broken ref: ', message selector
)
public printOn: stream = (
	stream nextPutAll: 'InternalBrokenRef: '.
	problem printOn: stream.
)
public refPrintOn: stream = (
	stream nextPutAll: '<broken ref: '.
	problem printOn: stream.
	stream nextPutAll: '>'.
)
public shorten ^<InternalRef> = (
	^self
)
public whenFulfilled: onValue whenBroken: onError ^<Ref> = (
	| localPromise = InternalLocalPromise __owner: __owner. |
	currentActor enqueueMessage:
		(PendingDelivery
			receiver: onError
			selector: #cull:
			arguments: {problem}
			resolver: localPromise).
	^localPromise externalRef
)
public wrapFor: targetActor = (
	self checkCurrentActor.
	#BOGUS.
	^(InternalBrokenRef problem: problem __owner: targetActor) externalRef
)
) : (
)
class InternalFarReference target: t targetsActor: ta __owner: o = InternalRef __owner: o (|
protected target = t.
protected targetsActor = ta.
|) (
public addReactor: reactor = (
	reactor reactResolved: externalRef
)
public checkCycle ^<Boolean> = (
	^false
)
public eventualSend: selector <Symbol> arguments: arguments <Array> ^<Ref> = (
	| resolverInTarget remotePromise redirector |
	resolverInTarget:: InternalLocalPromise __owner: targetsActor.
	targetsActor enqueueMessage:
		(PendingDelivery
			receiver: target
			selector: selector
			arguments: (wrapArguments: arguments from: currentActor to: targetsActor)
			resolver: resolverInTarget).

	remotePromise:: InternalRemotePromise
		target: resolverInTarget
		targetsActor: targetsActor
		__owner: currentActor.

	redirector:: RedirectReactor
		from: targetsActor
		to: currentActor
		resolver: remotePromise.
	targetsActor enqueueMessage:
		(PendingDelivery
			receiver: resolverInTarget
			selector: #addReactor:
			arguments: {redirector}
			resolver: nil).

	^remotePromise externalRef
)
public immediateSend: message <Message> ^<Object> = (
	^Error signal: 'Cannot immediate-send to an eventual ref: ', message selector
)
public printOn: stream = (
	stream nextPutAll: 'InternalFarRef: '.
	target printOn: stream.
)
public refPrintOn: stream = (
	stream nextPutAll: '<far ref: '.
	target printOn: stream.
	stream nextPutAll: '>'.
)
public shorten ^<InternalRef> = (
	^self
)
public whenFulfilled: onValue whenBroken: onError ^<Ref> = (
	| localPromise = InternalLocalPromise __owner: __owner. |
	currentActor enqueueMessage:
		(PendingDelivery
			receiver: onValue
			selector: #cull:
			arguments: {externalRef}
			resolver: localPromise).
	^localPromise externalRef
)
public wrapFor: targetActor = (
	targetsActor == targetActor ifTrue: [^target].
	^(InternalFarReference target: target targetsActor: targetsActor __owner: targetActor) externalRef.
)
) : (
)
class InternalLocalPromise __owner: o = InternalRef __owner: o (|
protected state ::= #unresolved. (* resolving. resolved *)
protected resolution <InternalRef>
protected reactors
|) (
public addReactor: reactor = (
	state == #unresolved ifTrue:
		[reactors == nil ifTrue: [reactors:: List new].
		reactors add: reactor.
		^self].
	state == #resolved ifTrue:
		[^resolution addReactor: reactor].
	halt.
)
public checkCycle ^<Boolean> = (
	state == #resolving ifTrue: [^true].
	state == #unresolved ifTrue: [^false].
	state == #resolved ifTrue: [^resolution checkCycle].
	halt.
)
public eventualSend: selector <Symbol> arguments: arguments <Array> ^<Ref> = (
	| localPromise |
	self checkCurrentActor.
	state == #resolved ifTrue: [^resolution eventualSend: selector arguments: arguments].
	localPromise:: InternalLocalPromise __owner: __owner.
	self addReactor: (EventualSendReactor selector: selector arguments: arguments resolver: localPromise).
	^localPromise externalRef
)
public immediateSend: message <Message> ^<Object> = (
	state == #resolved ifTrue: [^resolution immediateSend: message].
	^Error signal: 'Cannot immediate-send to an eventual ref: ', message selector
)
public internalResolve: internalRef <InternalRef> = (
	state == #unresolved ifFalse: [^Error signal: 'Already resolved'].
	internalRef isInternalRef ifFalse: [halt].

	state:: #resolving.
	internalRef checkCycle
		ifTrue: [resolution:: InternalBrokenRef problem: (Error new messageText: 'Cyclic resolution') __owner: __owner]
		ifFalse: [resolution:: internalRef].
	state:: #resolved.

	nil == reactors ifFalse:
		[reactors do: [:reactor | resolution addReactor: reactor].
		reactors:: nil].
)
public printOn: stream = (
	stream nextPutAll: 'InternalLocalPromise: '.
	state printOn: stream.
)
public refPrintOn: stream = (
	state == #resolved
		ifTrue: [resolution refPrintOn: stream]
		ifFalse: [stream nextPutAll: '<local promise>'].
)
public shorten ^<InternalRef> = (
	state == #unresolved ifTrue: [^self].
	state == #resolved ifTrue: [^resolution].
	halt.
)
public whenFulfilled: onValue whenBroken: onError ^<Ref> = (
	| localPromise |
	self checkCurrentActor.
	state == #resolved ifTrue: [^resolution whenFulfilled: onValue whenBroken: onError].
	localPromise:: InternalLocalPromise __owner: __owner.
	self addReactor: (WhenReactor onValue: onValue onError: onError resolver: localPromise).
	^localPromise externalRef
)
public wrapFor: targetActor = (
	state == #resolved ifTrue: [^resolution wrapFor: targetActor].

	checkCurrentActor.
	targetActor == __owner ifTrue: [halt].

	^(InternalRemotePromise target: self targetsActor: currentActor __owner: targetActor) externalRef
)
) : (
)
class InternalNearRef wrapping: t __owner: o = InternalRef __owner: o (|
protected target = t.
|) (
public addReactor: reactor = (
	reactor reactResolved: target.
)
public checkCycle ^<Boolean> = (
	^false
)
public eventualSend: selector <Symbol> arguments: arguments <Array> ^<Ref> = (
	| localPromise = InternalLocalPromise __owner: __owner. |
	currentActor enqueueMessage:
		(PendingDelivery
			receiver: target
			selector: selector
			arguments: arguments
			resolver: localPromise).
	^localPromise externalRef
)
public immediateSend: message <Message> ^<Object> = (
	^message sendTo: target
)
public printOn: stream = (
	stream nextPutAll: 'InternalNearRef: '.
	target printOn: stream.
)
public refPrintOn: stream = (
	stream nextPutAll: '<near ref: '.
	target printOn: stream.
	stream nextPutAll: '>'.
)
public shorten ^<InternalRef> = (
	^self
)
public whenFulfilled: onValue whenBroken: onError ^<Ref> = (
	| localPromise = InternalLocalPromise __owner: __owner. |
	currentActor enqueueMessage:
		(PendingDelivery
			receiver: onValue
			selector: #cull:
			arguments: {target}
			resolver: localPromise).
	^localPromise externalRef
)
public wrapFor: targetActor = (
	self checkCurrentActor.
	(isValueObject: target) ifTrue: [^target].
	^createFarReferenceTo: target in: currentActor for: targetActor
)
) : (
)
class InternalRef __owner: o = (|
public __owner <InternalActor> = o.
|) (
protected checkCurrentActor = (
	currentActor == __owner ifFalse: [halt].
)
public checkCycle ^<Boolean> = (
	subclassResponsibility
)
public eventualSend: selector <Symbol> arguments: arguments <Array> ^<Ref> = (
	subclassResponsibility
)
public externalRef = (
	^Ref wrapping: self
)
public immediateSend: message <Message> ^<Object> = (
	subclassResponsibility
)
public isInternalRef ^<Boolean> = (
	^true
)
public refPrintOn: stream = (
	subclassResponsibility
)
public shorten ^<InternalRef> = (
	subclassResponsibility
)
public whenFulfilled: onValue whenBroken: onError ^<Ref> = (
	subclassResponsibility
)
public wrapFor: targetActor = (
	subclassResponsibility
)
) : (
)
class InternalRemotePromise target: t targetsActor: ta __owner: o = InternalRef __owner: o (|
target <InternalLocalPromise> = t.
targetsActor <InternalActor> = ta.
state ::= #unresolved.
resolution <InternalRef>
|) (
public addReactor: reactor = (
	state == #resolved ifTrue: [^resolution addReactor: reactor].
	state == #unresolved ifTrue: [^reactor reactMoreResolved: self].
	stop5.
)
public checkCycle ^<Boolean> = (
	state == #resolving ifTrue: [^true].
	state == #unresolved ifTrue: [^false].
	state == #resolved ifTrue: [^resolution checkCycle].
	halt.
)
public eventualSend: selector <Symbol> arguments: arguments <Array> ^<Ref> = (
	| resolverInTarget remotePromise redirector |
	resolverInTarget:: InternalLocalPromise __owner: targetsActor.
	targetsActor enqueueMessage:
		(PendingDelivery
			receiver: target
			selector: #eventualSend:arguments:
			arguments: {selector. (wrapArguments: arguments from: currentActor to: targetsActor)}
			resolver: resolverInTarget).

	remotePromise:: InternalRemotePromise
		target: resolverInTarget
		targetsActor: targetsActor
		__owner: currentActor.

	redirector:: RedirectReactor
		from: targetsActor
		to: currentActor
		resolver: remotePromise.
	targetsActor enqueueMessage:
		(PendingDelivery
			receiver: resolverInTarget
			selector: #addReactor:
			arguments: {redirector}
			resolver: nil).

	^remotePromise externalRef
)
public immediateSend: message <Message> ^<Object> = (
	state == #resolved ifTrue: [^resolution immediateSend: message].
	^Error signal: 'Cannot immediate-send to an eventual ref: ', message selector
)
public internalResolve: internalRef <InternalRef> = (
	state == #unresolved ifFalse: [^Error signal: 'Already resolved'].
	internalRef isInternalRef ifFalse: [halt].

	state:: #resolving.
	internalRef checkCycle
		ifTrue: [resolution:: InternalBrokenRef problem: (Error new messageText: 'Cyclic resolution') __owner: __owner]
		ifFalse: [resolution:: internalRef].
	state:: #resolved.
)
public printOn: stream = (
	stream nextPutAll: 'InternalRemotePromise: '.
	state printOn: stream.
)
public redirectFrom: sourceActor to: destinationActor resolving: resolver = (
	| redirector |
	destinationActor == targetsActor ifTrue:
		[targetsActor enqueueMessage:
			(PendingDelivery
				receiver: resolver
				selector: #internalResolve:
				arguments: {target}
				resolver: nil).
		^self].

	redirector:: RedirectReactor
		from: targetsActor
		to: destinationActor
		resolver: resolver.
	targetsActor enqueueMessage:
		(PendingDelivery
			receiver: target
			selector: #addReactor:
			arguments: {redirector}
			resolver: nil).
)
public refPrintOn: stream = (
	state == #resolved
		ifTrue: [resolution refPrintOn: stream]
		ifFalse: [stream nextPutAll: '<remote promise>'].
)
public shorten ^<InternalRef> = (
	state == #unresolved ifTrue: [^self].
	state == #resolved ifTrue: [^resolution].
	halt.
)
public whenFulfilled: onValue whenBroken: onError ^<Ref> = (
	| localPromise redirector |
	self checkCurrentActor.
	state == #resolved ifTrue: [^resolution whenFulfilled: onValue whenBroken: onError].

	localPromise:: InternalLocalPromise __owner: __owner.
	redirector:: RedirectReactor
		from: targetsActor
		to: currentActor
		resolver: localPromise.
	targetsActor enqueueMessage:
		(PendingDelivery
			receiver: target
			selector: #addReactor:
			arguments: {redirector}
			resolver: nil).
	^localPromise whenFulfilled: onValue whenBroken: onError.
)
public wrapFor: targetActor = (
	state == #resolved ifTrue: [^resolution wrapFor: targetActor].

	targetActor == targetsActor ifTrue: [^target externalRef].

	^(InternalRemotePromise target: target targetsActor: targetsActor __owner: targetActor) externalRef
)
) : (
)
protected class PendingDelivery receiver: r selector: s arguments: a resolver: y = (
(* A message in an actor's event queue. Every turn starts by sending #deliverIn:.

Note that all slots contain objects that belong to the actor in which the message is queued, even the resolver. *)
|
protected receiver <Object> = r.
protected selector <Symbol> = s.
protected arguments <Array> = a.
protected resolver <InternalResolver> = y.
|) (
public deliverIn: fastCurrentActor = (
	| resolution type msg |

	(* Transcript print: fastCurrentActor; space; print: self; cr. *)

	[(* BOGUS: Don't rely on perform not being overridden. *)
	resolution:: receiver perform: selector withArguments: arguments.
	type:: #internalResolve:]
		on: Error (* BOGUS: Exception? *)
		do: [:error |
			resolver == nil ifTrue: [error pass].
			resolution:: error.
			type:: #internalBreak:].

	(* Transcript show: {resolver. type. resolution}; cr. *)

	nil == resolver ifTrue: [
		type == #internalBreak: ifTrue: [{'silent failure'. resolution} out].
		^self].

	resolver internalResolve: (type == #internalResolve:
		ifTrue: [asInternalRef: resolution]
		ifFalse: [InternalBrokenRef problem: resolution __owner: self]).
)
public postedTo: actor = (
	(*check: receiver allowedIn: actor.
	arguments do: [:arg | check: arg allowedIn: actor].*)

	selector numArgs == arguments size ifFalse: [halt].
	resolver == nil ifFalse: [resolver __owner == actor ifFalse: [halt]].
)
public printOn: stream = (
	receiver printOn: stream.
	stream nextPutAll: ' <-: '.
	stream nextPutAll: selector.
	stream space.
	arguments isEmpty ifFalse: [arguments printOn: stream].
)
) : (
)
protected class ProcessActor named: n = InternalActor (
(* An actor whose turns run in its own Process, allowing us to suspend it individually. *)
|
	protected name <String> = n.
	protected process <Process>
	protected messageQueue <SharedQueue> = SharedQueue2 new.
|) (
doOneTurn = (
	messageQueue next deliverIn: self.
)
public enqueueMessage: actorMessage = (
	actorMessage postedTo: self.
	messageQueue nextPut: actorMessage
)
public printOn: stream = (
	stream nextPutAll: '[ProcessActor('.
	stream nextPutAll: name.
	stream nextPutAll: ')#'.
	hash printOn: stream.
	stream nextPutAll: ']'.
)
public startEventLoop = (
	process:: [[self doOneTurn] repeat] newProcess.
	process name: ('Event loop for ', name) asString.
	process priority: Processor userBackgroundPriority.
	process actor: self.
	process resume.
)
) : (
)
class PromiseUtils = () (
public all: refs <List[Promise[V, E]]> ^<Promise[List[V], E]> = (
	(* Answers a promise that resolves with all the resolutions of refs, or the first problem. *)
)
public broken: problem <E> ^<Promise[nil, E]> = (
	^Resolver new break: problem; promise
)
public fulfilled: resolution <V> ^<Promise[V, nil]> = (
	^Resolver new fulfill: resolution; promise
)
public timeout: ref <Promise[V, E]> after: milliseconds <Integer> ^<Promise[V, E | TimeoutError]> = (
	(* Answers a promise that resolves with the resolution of ref, or a TimeoutError if refs is not resolved within milliseconds. *)
	#BOGUS. (* Should have proper duration type. *)
)
public when: ref <Promise[V, E]> fulfilled: onValue <[:V | V2 def]> ^<Promise[V2, E]> = (
	(* Subscribe to notification of a reference being fulfilled. If ref is fulfilled, onValue will run in some turn after the turn that resolves ref, and the returned promise with be resolved with the result of onValue. If ref is broken, the returned promise with be broken with the same error. *)
	^(asInternalRef: ref) whenFulfilled: onValue whenBroken: [:e | e signal] (* Should we make this the behavior of a nil catch-clause instead? *)
)
public when: ref <Promise[V, E]> fulfilled: onValue <[:V | V2 def]> broken: onError <[:E | E2 def]> ^<Promise[V2, E2]> = (
	(* Subscribe to notification of a reference being fulfilled or broken. onValue or onError respectively will run in some turn after the turn that resolves ref., and the returned promise with be resolved with the result of the callback. *)
	^(asInternalRef: ref) whenFulfilled: onValue whenBroken: onError
)
public when: ref <Promise> partitioned: onPartition <[:Promise | R def] | [R def]> ^<Promise[R, nil]> = (
	(* Subscribe to notification of a reference breaking due to partition. *)
)
) : (
)
protected class RedirectReactor from: s to: d resolver: r = (
(* A bridge between InternalResolvers in different actors. *)
|
protected sourceActor <Actor> = s.
protected destinationActor <Actor> = d.
protected resolver <InternalResolver> = r.
|) (
public reactBroken: r = (
	| msg |
	currentActor == sourceActor ifFalse: [halt].
	msg:: PendingDelivery
		receiver: resolver
		selector: #internalResolve:
		arguments: {InternalBrokenRef problem: r __owner: destinationActor}
		resolver: nil.
	destinationActor enqueueMessage: msg.
)
public reactMoreResolved: internalRef = (
	internalRef redirectFrom: sourceActor to: destinationActor resolving: resolver.
)
public reactResolved: r = (
	| msg |
	currentActor == sourceActor ifFalse: [halt].
	msg:: PendingDelivery
		receiver: resolver
		selector: #internalResolve:
		arguments: {asInternalRef: (wrapArgument: r from: sourceActor to: destinationActor)}
		resolver: nil.
	destinationActor enqueueMessage: msg.
)
) : (
)
class Ref wrapping: ref = (|
protected internalRef = ref.
|internalRefs at: self put: ref) (
protected doesNotUnderstand: message <Message> = (
	(* For the debugger/inspector. Questionable. *)
	(* Should probably override public method of Object as well. *)
	0 = message arguments size ifTrue: [(message selector startsWith: 'is') ifTrue: [^false]].
	^internalRef immediateSend: message
)
public isPromise = (
	^true
)
public printOn: stream = (
	(* BOGUS - debugging *)
	internalRef refPrintOn: stream.
)
protected yourself = (
	^super yourself
)
) : (
)
public class Resolver = (
(*A Resolver represents the authority to resolve a Promise and is the control facet of the promise's eventual resolution.*)
|
	private (* N.B. We must not leak access to an InternalResolver via subobjects. *) internalResolver = InternalLocalPromise __owner: currentActor.
	public promise <Promise[V, E]> = internalResolver externalRef.
|) (
public break: error <E> = (
	internalResolver internalResolve: (InternalBrokenRef problem: error __owner: currentActor).
)
public fulfill: value <V> = (
	internalResolver internalResolve: (asInternalRef: value).
)
public isResolver ^<Boolean> = (
	^true
)
) : (
)
public class Timer __duration: milliseconds callback: callback repeating: repeating = (|
	protected process
	protected actor = currentActor.
	protected cancelled ::= false.
|repeating
		ifTrue: [self schedule: callback every: milliseconds]
		ifFalse: [self schedule: callback after: milliseconds]) (
public cancel = (
	cancelled:: true.
	nil == process ifFalse: [process terminate. process:: nil].
)
public isActive ^<Boolean> = (
	cancelled ifTrue: [^false].
	^nil ~= process
)
public isTimer ^<Boolean> = (
	^true
)
protected schedule: callback after: ms = (
	|
	delay = SqueakDelay forMilliseconds: ms.
	msg = PendingDelivery
		receiver: [cancelled ifFalse: [callback value]]
		selector: #value
		arguments: {}
		resolver: nil.
	|
	process:: [delay wait. actor enqueueMessage: msg. process:: nil] newProcess.
	process name: 'One-shot timer for ', callback printString.
	process priority: 41.
	process resume.
)
protected schedule: callback every: ms = (
	(* When should we start ticking down for the next callback? When we schedule the callback, when the callback starts, or if-and-when the callback completes without an error? *)
	|
	delay = SqueakDelay forMilliseconds: ms.
	msg = PendingDelivery
		receiver: [cancelled ifFalse: [callback value: self]]
		selector: #value
		arguments: {}
		resolver: nil.
	|
	process:: [[delay wait. actor enqueueMessage: msg] repeat] newProcess.
	process name: 'Repeating timer for ', callback printString.
	process priority: 41.
	process resume.
)
) : (
public after: duration do: callback <[]> ^<Timer> = (
	^self __duration: duration callback: callback repeating: false
)
public every: duration do: callback <[:Timer]> ^<Timer> = (
	^self __duration: duration callback: callback repeating: true
)
)
protected class UIActor = InternalActor (
(* An actor whose turns run in the Squeak UI Process. *)
) (
public enqueueMessage: actorMessage = (
	actorMessage postedTo: self.

	(* bogus - should drain an explicit queue to allow for a randomizing scheduler *)
	brazil theDesktop scheduleUIAction:
		[[actorMessage deliverIn: self] on: UnhandledError do: [:ignore | ]].
)
public printOn: stream = (
	stream nextPutAll: '[UIActor#'.
	hash printOn: stream.
	stream nextPutAll: ']'.
)
) : (
)
protected class WhenReactor onValue: v onError: e resolver: r = (
(* A when-catch for a promise.

Note that all slots contain objects that belong to the actor which sent #whenResolved:. *)
|
	protected onValue <[:V] | []> = v.
	protected onError <[:E] | []> = e.
	protected resolver <InternalResolver> = r.
|) (
public reactBroken: value = (
	resolver __owner == currentActor ifFalse: [halt].
	currentActor enqueueMessage:
		(PendingDelivery
			receiver: onError
			selector: #cull:
			arguments: {value}
			resolver: resolver).
)
public reactMoreResolved: internalRef = (
	resolver internalResolve: (asInternalRef: (internalRef whenFulfilled: onValue whenBroken: onError)).
)
public reactResolved: value = (
	resolver __owner == currentActor ifFalse: [halt].
	currentActor enqueueMessage:
		(PendingDelivery
			receiver: onValue
			selector: #cull:
			arguments: {value}
			resolver: resolver).
)
) : (
)
private asInternalRef: externalRefOrUnboxedNearRef = (
	^(isRef: externalRefOrUnboxedNearRef)
		ifTrue: [internalRefs at: externalRefOrUnboxedNearRef ifAbsent: [halt]]
		ifFalse: [InternalNearRef wrapping: externalRefOrUnboxedNearRef __owner: currentActor]
)
private check: object allowedIn: actor = (
	assert: [nil ~= actor] message: 'Invalid actor'.

	(isRef: object) ifTrue: [object __owner == actor ifFalse: [halt]].
	(* Otherwise near reference *)
)
private createFarReferenceTo: target in: targetActor for: sourceActor = (
	^(InternalFarReference target: target targetsActor: targetActor __owner: sourceActor) externalRef.
)
private createRemotePromiseOf: resolverInTarget in: targetActor for: sourceActor = (
	| sourceResolver redirector message remotePromise |
	sourceActor == targetActor ifTrue: [halt].

	sourceResolver:: InternalLocalPromise __owner: sourceActor.
	sourceResolver state: #remote.
	sourceResolver _targetPromise: resolverInTarget.
	sourceResolver _targetActor: targetActor.

	redirector:: RedirectReactor
		from: targetActor
		to: sourceActor
		resolver: sourceResolver.

	message:: PendingDelivery
		receiver: resolverInTarget
		selector: #addReactor:
		arguments: {redirector}
		resolver: nil.
	targetActor enqueueMessage: message.

	^Ref wrapping: sourceResolver __owner: sourceActor.
)
private currentActor = (
	^Processor activeProcess actor ifNil: [uiActor].
)
private isRef: object <Object> ^<Boolean> = (
	^Ref == (primitives classOf: object)
)
private isValueObject: object <Object> ^<Boolean> = (
	nil == object ifTrue: [^true].
	true == object ifTrue: [^true].
	false == object ifTrue: [^true].
	object isKindOfString ifTrue: [^true].
	object isKindOfNumber ifTrue: [^true].
	object isCharacter ifTrue: [^true].
	object isKindOfBehavior ifTrue: [^isValueObject: object enclosingClass].
	Promise == object ifTrue: [^true].
	^false
)
private wrapArgument: argument from: sourceActor to: targetActor = (
	(* [argument] lives in [sourceActor], answer the corresponding proxy that lives in [targetActor] *)

	sourceActor == targetActor ifTrue: [halt. (* We can't tell which side a near reference belongs to here, so we must have decided this earlier. *) ].

	^(asInternalRef: argument) wrapFor: targetActor
)
private wrapArguments: arguments from: sourceActor to: targetActor = (
	^arguments collect: [:argument | wrapArgument: argument from: sourceActor to: targetActor].
)
) : (
)