Skip to content

Commit 710c2a6

Browse files
rsshrssh
committed
implemented swothcing values in effected input
1 parent 86aa733 commit 710c2a6

File tree

8 files changed

+275
-213
lines changed

8 files changed

+275
-213
lines changed

‎src/main/scala/gopher/Transputer.scala

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -361,7 +361,7 @@ trait SelectTransputer extends ForeverSelectorBuilder with Transputer
361361
/**
362362
* configure loop in selector
363363
*/
364-
def loop(f: PartialFunction[Any,Unit]): Unit = macro SelectorBuilder.loopImpl[Unit]
364+
def loop(f: PartialFunction[Any,Unit]): Unit = macro SelectorBuilderImpl.loop[Unit]
365365

366366

367367
def stop():Unit = stopFlowTermination()

‎src/main/scala/gopher/channels/Continuated.scala

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ case class ContRead[A,B](
3232
override val flowTermination: FlowTermination[B]) extends FlowContinuated[B]
3333
{
3434
type El = A
35+
type F = ContRead[A,B]=>Option[ContRead.In[A] => Future[Continuated[B]]]
3536
}
3637

3738
object ContRead
@@ -43,6 +44,7 @@ object ContRead
4344
case object ChannelClosed extends In[Nothing]
4445
case class Failure(ex:Throwable) extends In[Nothing]
4546

47+
4648
object In
4749
{
4850
def value[A](a:A) = ContRead.Value(a)
@@ -84,6 +86,8 @@ object ContRead
8486
type S=B
8587
type F = ContRead[A,B]=>Option[ContRead.In[A]=>Future[Continuated[B]]]
8688
}
89+
90+
type AuxF[A,B] = ContRead[A,B]=>Option[ContRead.In[A]=>Future[Continuated[B]]]
8791
}
8892

8993

‎src/main/scala/gopher/channels/FoldSelectorBuilder.scala

Lines changed: 87 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,8 @@ import gopher.util._
88
import scala.concurrent._
99
import scala.concurrent.duration._
1010
import scala.annotation.unchecked._
11-
11+
import java.util.{WeakHashMap=>JWeakHashMap}
12+
import java.util.function.{BiConsumer=>JBiConsumer}
1213

1314

1415
trait FoldSelectorBuilder[T] extends SelectorBuilder[T]
@@ -68,17 +69,53 @@ class FoldSelect[T](sf:SelectFactory) extends FoldSelectorBuilder[T]
6869
class FoldSelectorEffectedInput[A](s:()=>Input[A],val api: GopherAPI) extends Input[A]
6970
{
7071

72+
thisInput =>
73+
7174
def current = s()
72-
75+
val activeReaders = new JWeakHashMap[Reader[_],JWeakHashMap[Input[A],Boolean]]
76+
77+
class Reader[B](val svInput: Input[A],
78+
val function: ContRead[A,B]=>Option[ContRead.In[A]=>Future[Continuated[B]]],
79+
val flowTermination: FlowTermination[B], var disabled: Boolean=false)
80+
extends (ContRead[A,B]=>Option[ContRead.In[A] => Future[Continuated[B]]]) {
81+
def apply(cr:ContRead[A,B]):Option[ContRead.In[A]=>Future[Continuated[B]]] =
82+
if (disabled || !(svInput eq s())) None else function(cr.copy(channel=thisInput))
83+
84+
85+
def cbread(ch:Input[A]):Unit =
86+
ch.cbread(this,flowTermination)
87+
}
88+
7389
def cbread[B](f: ContRead[A,B] => Option[ContRead.In[A] => Future[Continuated[B]]],ft: FlowTermination[B]): Unit = {
74-
val sv = s()
75-
sv.cbread[B](cr=>if (sv eq s()) f(cr.copy(channel=this)) else None ,ft)
90+
val r = new Reader[B](current,f,ft,false)
91+
val s = current
92+
val jwhashMap = new JWeakHashMap[Input[A],Boolean]
93+
jwhashMap.put(current,true)
94+
activeReaders.put(r,jwhashMap)
95+
current.cbread(r,ft)
96+
}
97+
98+
def refresh():Unit =
99+
{
100+
activeReaders forEach refreshBiConsumer
101+
}
102+
103+
val refreshBiConsumer = new JBiConsumer[Reader[_],JWeakHashMap[Input[A],Boolean]] {
104+
override def accept(r:Reader[_], sources:JWeakHashMap[Input[A],Boolean]):Unit =
105+
{
106+
val s = current
107+
val alreadyRegistered = sources.containsKey(s)
108+
if (!alreadyRegistered) {
109+
sources.put(s,true)
110+
r.cbread(s)
111+
}
112+
}
76113
}
77114

78115

79116
}
80117

81-
class FoldSelectorBuilderImpl(val c:Context)
118+
class FoldSelectorBuilderImpl(override val c:Context) extends SelectorBuilderImpl(c)
82119
{
83120
import c.universe._
84121

@@ -163,9 +200,6 @@ class FoldSelectorBuilderImpl(val c:Context)
163200
c.Expr[Future[S]](tree)
164201
}
165202

166-
def transformSelectMatch(bn:TermName,cases: List[CaseDef]):List[Tree]=
167-
SelectorBuilder.transformSelectMatch(c)(bn,cases)
168-
169203
def fold[S:c.WeakTypeTag](s:c.Expr[S])(op:c.Expr[(S,FoldSelect[S])=>S]):c.Expr[S] =
170204
c.Expr[S](q"scala.async.Async.await(${afold(s)(op).tree})")
171205

@@ -340,7 +374,7 @@ class FoldSelectorBuilderImpl(val c:Context)
340374
guard,
341375
Match(Ident(choice1),cases1)) =>
342376
if (sel == choice1) {
343-
val selectSymbols = retrieveSelectChannels(cases)
377+
val selectSymbols = retrieveSelectChannels(cases1)
344378
FoldParse(
345379
stateVal = x,
346380
stateUsedInSelect = selectSymbols.contains(x.symbol),
@@ -386,8 +420,50 @@ class FoldSelectorBuilderImpl(val c:Context)
386420

387421
private def retrieveSelectChannels(cases:List[CaseDef]): Set[c.Symbol] =
388422
{
389-
//TODO: implement
390-
Set()
423+
val s0=Set[c.Symbol]()
424+
cases.foldLeft(s0){ (s,e) =>
425+
def addSymbol(in:Tree) = s+e.symbol
426+
acceptSelectCaseDefPattern(e, addSymbol, addSymbol, addSymbol, _ => s)
427+
}
428+
}
429+
430+
//TODO: generalize and merge with parsing in SelectorBuilderImpl
431+
def acceptSelectCaseDefPattern[A](caseDef:CaseDef,onRead: Tree => A, onWrite: Tree => A,
432+
onSelectTimeout: Tree => A, onIdle: Tree => A):A =
433+
{
434+
caseDef.pat match {
435+
case Bind(name,t) =>
436+
val termName = name.toTermName
437+
t match {
438+
case Typed(_,tp:TypeTree) =>
439+
val tpoa = if (tp.original.isEmpty) tp else tp.original
440+
val tpo = MacroUtil.skipAnnotation(c)(tpoa)
441+
tpo match {
442+
case Select(ch,TypeName("read")) => onRead(ch)
443+
case Select(ch,TypeName("write")) => onWrite(ch)
444+
case Select(select,TypeName("timeout")) => onSelectTimeout(select)
445+
case _ =>
446+
if (caseDef.guard.isEmpty) {
447+
c.abort(tp.pos, "row caseDef:"+showRaw(caseDef) );
448+
c.abort(tp.pos, "match pattern in select without guard must be in form x:channel.write or x:channel.read");
449+
} else {
450+
parseGuardInSelectorCaseDef(termName, caseDef.guard) match {
451+
case q"scala.async.Async.await[${t}](${readed}.aread):${t1}" =>
452+
onRead(readed)
453+
case q"scala.async.Async.await[${t}](${ch}.awrite($expression)):${t1}" =>
454+
onWrite(ch)
455+
case x@_ =>
456+
c.abort(tp.pos, "can't parse match guard: "+x);
457+
}
458+
}
459+
}
460+
case _ =>
461+
c.abort(caseDef.pat.pos,"x:channel.read or x:channel.write form is required")
462+
}
463+
case Ident(TermName("_")) => onIdle(caseDef.pat)
464+
case _ =>
465+
c.abort(caseDef.pat.pos,"bind in pattern is expected")
466+
}
391467
}
392468

393469
}

‎src/main/scala/gopher/channels/ForeverSelectorBuilder.scala

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,7 @@ trait ForeverSelectorBuilder extends SelectorBuilder[Unit]
6969
* Note, that you can use implicit instance of [FlowTermination[Unit]] to stop loop.
7070
**/
7171
def foreach(f:Any=>Unit):Unit =
72-
macro SelectorBuilder.foreachImpl[Unit]
72+
macro SelectorBuilderImpl.foreach[Unit]
7373

7474

7575

@@ -83,7 +83,7 @@ trait ForeverSelectorBuilder extends SelectorBuilder[Unit]
8383
*}}}
8484
*/
8585
def apply(f: PartialFunction[Any,Unit]): Future[Unit] =
86-
macro SelectorBuilder.applyImpl[Unit]
86+
macro SelectorBuilderImpl.apply[Unit]
8787

8888

8989
def inputBuilder[B]() = new InputSelectorBuilder[B](api)
@@ -100,10 +100,10 @@ trait ForeverSelectorBuilder extends SelectorBuilder[Unit]
100100
*
101101
*}}}
102102
**/
103-
def map[B](f:Any=>B):Input[B] = macro SelectorBuilder.mapImpl[B]
103+
def map[B](f:Any=>B):Input[B] = macro SelectorBuilderImpl.map[B]
104104

105105
def input[B](f:PartialFunction[Any,B]):Input[B] =
106-
macro SelectorBuilder.inputImpl[B]
106+
macro SelectorBuilderImpl.input[B]
107107

108108
}
109109

‎src/main/scala/gopher/channels/InputSelectorBuilder.scala

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -68,10 +68,10 @@ class InputSelectorBuilder[T](override val api: GopherAPI) extends SelectorBuild
6868

6969

7070
def foreach(f:Any=>T):T =
71-
macro SelectorBuilder.foreachImpl[T]
71+
macro SelectorBuilderImpl.foreach[T]
7272

7373
def apply(f: PartialFunction[Any,T]): Future[T] =
74-
macro SelectorBuilder.applyImpl[T]
74+
macro SelectorBuilderImpl.apply[T]
7575

7676
// input methods
7777
def cbread[B](f:

‎src/main/scala/gopher/channels/OnceSelectorBuilder.scala

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,10 +51,10 @@ trait OnceSelectorBuilder[T] extends SelectorBuilder[T@uncheckedVariance]
5151

5252

5353
def foreach(f:Any=>T):T =
54-
macro SelectorBuilder.foreachImpl[T]
54+
macro SelectorBuilderImpl.foreach[T]
5555

5656
def apply(f: PartialFunction[Any,T]): Future[T] =
57-
macro SelectorBuilder.applyImpl[T]
57+
macro SelectorBuilderImpl.apply[T]
5858

5959
}
6060

0 commit comments

Comments
 (0)