209 Push to stream (#214)

aqua-src/test.aqua

@@ -1,8 +1,9 @@
-service AquaDHT("aqua-dht"):
+service OpH("oph"):
-  put_host_value(key: string, value: string, service_id: []string)
+  get_str() -> string
 
-func putHostValue(key: string, value: string, service_id: ?string):
+func create_client_util() -> []string:
-    AquaDHT.put_host_value(key, value, service_id)
+    results: *string
-
+    results <<- "hello"
-func create_client_util(service_id: string):
+    str <- OpH.get_str()
-    putHostValue("client-util", service_id, nil)
+    results <<- str
+    <- results

build.sbt

@@ -28,7 +28,7 @@ val cats = "org.typelevel"             %% "cats-core"    % catsV
 name := "aqua-hll"
 
 val commons = Seq(
-  baseAquaVersion := "0.1.9",
+  baseAquaVersion := "0.1.10",
   version         := baseAquaVersion.value + "-" + sys.env.getOrElse("BUILD_NUMBER", "SNAPSHOT"),
   scalaVersion    := dottyVersion,
   libraryDependencies ++= Seq(

parser/src/main/scala/aqua/parser/expr/ForExpr.scala

@@ -26,6 +26,7 @@ object ForExpr extends Expr.AndIndented {
       CallArrowExpr ::
       AbilityIdExpr ::
       AssignmentExpr ::
+      PushToStreamExpr ::
       Expr.defer(TryExpr) ::
       Expr.defer(IfExpr) ::
       Expr.defer(ElseOtherwiseExpr) ::

parser/src/main/scala/aqua/parser/expr/FuncExpr.scala

@@ -21,6 +21,7 @@ object FuncExpr extends Expr.AndIndented {
   override def validChildren: List[Expr.Lexem] =
     AbilityIdExpr ::
       AssignmentExpr ::
+      PushToStreamExpr ::
       ReturnExpr ::
       ForExpr ::
       Expr.defer(OnExpr) ::

parser/src/main/scala/aqua/parser/expr/IfExpr.scala

@@ -18,6 +18,7 @@ object IfExpr extends Expr.AndIndented {
       CallArrowExpr ::
       AbilityIdExpr ::
       AssignmentExpr ::
+      PushToStreamExpr ::
       Expr.defer(ParExpr) ::
       Expr.defer(CoExpr) ::
       Expr.defer(TryExpr) ::

parser/src/main/scala/aqua/parser/expr/OnExpr.scala

@@ -16,6 +16,7 @@ object OnExpr extends Expr.AndIndented {
       CallArrowExpr ::
       AbilityIdExpr ::
       AssignmentExpr ::
+      PushToStreamExpr ::
       ParExpr ::
       CoExpr ::
       Expr.defer(TryExpr) ::

parser/src/main/scala/aqua/parser/expr/PushToStreamExpr.scala

@@ -0,0 +1,21 @@
+package aqua.parser.expr
+
+import aqua.parser.Expr
+import aqua.parser.lexer.Token._
+import aqua.parser.lexer.{Name, Value}
+import aqua.parser.lift.LiftParser
+import cats.Comonad
+import cats.parse.{Parser => P}
+
+case class PushToStreamExpr[F[_]](
+  stream: Name[F],
+  value: Value[F]
+) extends Expr[F](PushToStreamExpr, stream)
+
+object PushToStreamExpr extends Expr.Leaf {
+
+  override def p[F[_]: LiftParser: Comonad]: P[PushToStreamExpr[F]] =
+    ((Name.p[F] <* ` <<- `).with1 ~ Value.`value`).map { case (variable, value) =>
+      PushToStreamExpr(variable, value)
+    }
+}

parser/src/main/scala/aqua/parser/lexer/Token.scala

@@ -74,6 +74,7 @@ object Token {
   val ` -> ` : P[Unit] = P.string("->").surroundedBy(` `.?)
   val ` <- ` : P[Unit] = P.string("<-").surroundedBy(` `.?)
   val `=` : P[Unit] = P.string("=")
+  val ` <<- ` : P[Unit] = P.string("<<-").surroundedBy(` `.?)
   val ` = ` : P[Unit] = P.string("=").surroundedBy(` `.?)
   val `?` : P[Unit] = P.string("?")
   val `<-` : P[Unit] = P.string("<-")

parser/src/test/scala/aqua/AquaSpec.scala

@@ -72,6 +72,9 @@ trait AquaSpec extends EitherValues {
   def parseAssign(str: String): AssignmentExpr[Id] =
     AssignmentExpr.p[Id].parseAll(str).value
 
+  def parsePush(str: String): PushToStreamExpr[Id] =
+    PushToStreamExpr.p[Id].parseAll(str).value
+
   def parseConstant(str: String): ConstantExpr[Id] =
     ConstantExpr.p[Id].parseAll(str).value
 

parser/src/test/scala/aqua/parser/PushToStreamExprSpec.scala

@@ -0,0 +1,21 @@
+package aqua.parser
+
+import aqua.AquaSpec
+import aqua.parser.expr.PushToStreamExpr
+import cats.Id
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class PushToStreamExprSpec extends AnyFlatSpec with Matchers with AquaSpec {
+  import AquaSpec._
+
+  "assign" should "be parsed" in {
+    parsePush("a <<- \"b\"") should be(
+      PushToStreamExpr[Id]("a", toStr("b"))
+    )
+
+    parsePush("a <<- b") should be(
+      PushToStreamExpr[Id]("a", toVar("b"))
+    )
+  }
+}

semantics/src/main/scala/aqua/semantics/ExprSem.scala

@@ -20,6 +20,7 @@ object ExprSem {
     expr match {
       case expr: AbilityIdExpr[F] => new AbilityIdSem(expr).program[G]
       case expr: AssignmentExpr[F] => new AssignmentSem(expr).program[G]
+      case expr: PushToStreamExpr[F] => new PushToStreamSem(expr).program[G]
       case expr: AliasExpr[F] => new AliasSem(expr).program[G]
       case expr: ConstantExpr[F] => new ConstantSem(expr).program[G]
       case expr: DeclareStreamExpr[F] => new DeclareStreamSem(expr).program[G]

semantics/src/main/scala/aqua/semantics/expr/AssignmentSem.scala

@@ -1,14 +1,12 @@
 package aqua.semantics.expr
 
 import aqua.model.Model
-import aqua.model.func.raw.{AssignmentTag, FuncOp, FuncOps}
+import aqua.model.func.raw.{AssignmentTag, FuncOp}
-import aqua.parser.expr.{AbilityIdExpr, AssignmentExpr}
+import aqua.parser.expr.AssignmentExpr
 import aqua.semantics.Prog
 import aqua.semantics.rules.ValuesAlgebra
-import aqua.semantics.rules.abilities.AbilitiesAlgebra
 import aqua.semantics.rules.names.NamesAlgebra
 import cats.free.Free
-import cats.syntax.flatMap._
 import cats.syntax.functor._
 
 class AssignmentSem[F[_]](val expr: AssignmentExpr[F]) extends AnyVal {

semantics/src/main/scala/aqua/semantics/expr/OnSem.scala

@@ -1,7 +1,7 @@
 package aqua.semantics.expr
 
-import aqua.model.{Model, ValueModel}
 import aqua.model.func.raw.{FuncOp, OnTag}
+import aqua.model.{Model, ValueModel}
 import aqua.parser.expr.OnExpr
 import aqua.semantics.Prog
 import aqua.semantics.rules.ValuesAlgebra
@@ -12,7 +12,6 @@ import cats.Traverse
 import cats.data.Chain
 import cats.free.Free
 import cats.syntax.apply._
-import cats.syntax.functor._
 import cats.syntax.flatMap._
 
 class OnSem[F[_]](val expr: OnExpr[F]) extends AnyVal {

semantics/src/main/scala/aqua/semantics/expr/PushToStreamSem.scala

@@ -0,0 +1,73 @@
+package aqua.semantics.expr
+
+import aqua.model.func.Call
+import aqua.model.func.raw.{CallServiceTag, FuncOp}
+import aqua.model.{LiteralModel, Model}
+import aqua.parser.expr.PushToStreamExpr
+import aqua.parser.lexer.Token
+import aqua.semantics.Prog
+import aqua.semantics.rules.ValuesAlgebra
+import aqua.semantics.rules.names.NamesAlgebra
+import aqua.semantics.rules.types.TypesAlgebra
+import aqua.types.{StreamType, Type}
+import cats.free.Free
+import cats.syntax.apply._
+
+class PushToStreamSem[F[_]](val expr: PushToStreamExpr[F]) extends AnyVal {
+
+  private def ensureStreamElementMatches[Alg[_]](
+    streamToken: Token[F],
+    elementToken: Token[F],
+    streamOp: Option[Type],
+    elementOp: Option[Type]
+  )(implicit
+    T: TypesAlgebra[F, Alg]
+  ): Free[Alg, Boolean] =
+    (streamOp, elementOp).mapN { case (stream, element) =>
+      stream match {
+        case StreamType(st) =>
+          T.ensureTypeMatches(elementToken, st, element)
+        case _ =>
+          T.ensureTypeMatches(streamToken, StreamType(element), stream)
+      }
+
+    }.getOrElse(Free.pure[Alg, Boolean](false))
+
+  def program[Alg[_]](implicit
+    N: NamesAlgebra[F, Alg],
+    T: TypesAlgebra[F, Alg],
+    V: ValuesAlgebra[F, Alg]
+  ): Prog[Alg, Model] =
+    V.valueToModel(expr.value).flatMap {
+      case Some(vm) =>
+        for {
+          resolvedStreamTypeOp <- N.read(expr.stream)
+          valueType <- V.resolveType(expr.value)
+          ensure <- ensureStreamElementMatches(
+            expr.token,
+            expr.value,
+            resolvedStreamTypeOp,
+            valueType
+          )
+        } yield {
+          if (ensure)
+            resolvedStreamTypeOp
+              .map(t =>
+                FuncOp
+                  .leaf(
+                    CallServiceTag(
+                      LiteralModel.quote("op"),
+                      "identity",
+                      Call(vm :: Nil, Some(Call.Export(expr.stream.value, t)))
+                    )
+                  ): Model
+              )
+              .getOrElse(Model.error("Cannot resolve stream type"))
+          else
+            Model.error("Stream and pushed element types are not matches")
+        }
+
+      case _ => Free.pure[Alg, Model](Model.error("Cannot resolve value"))
+    }
+
+}