Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Onur Gumus
committed
Dec 7, 2019
1 parent
401cae3
commit a57252d
Showing
17 changed files
with
513 additions
and
44 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,10 +1,13 @@ | ||
| FSharp.Core | ||
| Microsoft.EntityFrameworkCore | ||
| Microsoft.EntityFrameworkCore.Sqlite | ||
| Bolero | ||
| Bolero.Server | ||
| Bolero | ||
| Bolero.Server | ||
| Bolero.HotReload.Server | ||
| Microsoft.AspNetCore.Blazor | ||
| Microsoft.AspNetCore.Blazor.Server | ||
| Microsoft.AspNetCore.Blazor | ||
| Microsoft.AspNetCore.Blazor.Server | ||
| Microsoft.AspNetCore.Mvc.Razor.RuntimeCompilation | ||
| JWT | ||
| Akka.Cluster.Sharding | ||
| Akkling | ||
| Akka.Persistence.Sqlite |
Binary file not shown.
Binary file not shown.
Binary file not shown.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,144 @@ | ||
| module Actor | ||
|
|
||
| open Akka.Persistence.Sqlite | ||
| open Akkling | ||
| open Akka.Cluster.Tools | ||
| open Akka.Cluster.Tools.Singleton | ||
| open Akkling.Persistence | ||
| open FBlazorShop.App.Model | ||
| open Akka.Persistence.Query | ||
| open Akka.Persistence.Query.Sql | ||
| open Akka.Streams | ||
| open Akka.Persistence.Journal | ||
| open System.Collections.Immutable | ||
|
|
||
|
|
||
| let configWithPort port = | ||
| let config = Configuration.parse (""" | ||
| akka { | ||
| actor { | ||
| provider = "Akka.Cluster.ClusterActorRefProvider, Akka.Cluster" | ||
| serializers { | ||
| hyperion = "Akka.Serialization.HyperionSerializer, Akka.Serialization.Hyperion" | ||
| } | ||
| serialization-bindings { | ||
| // "System.Object" = hyperion | ||
| } | ||
| } | ||
| remote { | ||
| helios.tcp { | ||
| public-hostname = "localhost" | ||
| hostname = "localhost" | ||
| port = """ + port.ToString() + """ | ||
| } | ||
| } | ||
| cluster { | ||
| auto-down-unreachable-after = 5s | ||
| // seed-nodes = [ "akka.tcp://cluster-system@localhost:12345" ] | ||
| // sharding.remember-entities = true | ||
| } | ||
| persistence{ | ||
| query.journal.sql { | ||
| # Implementation class of the SQL ReadJournalProvider | ||
| class = "Akka.Persistence.Query.Sql.SqlReadJournalProvider, Akka.Persistence.Query.Sql" | ||
| # Absolute path to the write journal plugin configuration entry that this | ||
| # query journal will connect to. | ||
| # If undefined (or "") it will connect to the default journal as specified by the | ||
| # akka.persistence.journal.plugin property. | ||
| write-plugin = "" | ||
| # The SQL write journal is notifying the query side as soon as things | ||
| # are persisted, but for efficiency reasons the query side retrieves the events | ||
| # in batches that sometimes can be delayed up to the configured `refresh-interval`. | ||
| refresh-interval = 1s | ||
| # How many events to fetch in one query (replay) and keep buffered until they | ||
| # are delivered downstreams. | ||
| max-buffer-size = 100 | ||
| } | ||
| journal { | ||
| plugin = "akka.persistence.journal.sqlite" | ||
| sqlite | ||
| { | ||
| connection-string = "Data Source=pizza.db;" | ||
| auto-initialize = on | ||
| event-adapters.tagger = "Actor+Tagger, FBlazorShop" | ||
| event-adapter-bindings { | ||
| "Actor+Message, FBlazorShop" = tagger | ||
| } | ||
| } | ||
| } | ||
| snapshot-store{ | ||
| plugin = "akka.persistence.snapshot-store.sqlite" | ||
| sqlite { | ||
| auto-initialize = on | ||
| connection-string = "Data Source=pizza.db" | ||
| } | ||
| } | ||
| } | ||
| } | ||
| """) | ||
| config.WithFallback(ClusterSingletonManager.DefaultConfig()) | ||
|
|
||
| type Command = PlaceOrder of Order | ||
| type Event = OrderPlaced of Order | ||
|
|
||
| type Message = | ||
| | Command of Command | ||
| | Event of Event | ||
|
|
||
| let deft = ImmutableHashSet.Create("default") | ||
|
|
||
| type Tagger () = | ||
| interface IWriteEventAdapter with | ||
| member _.Manifest _ = "" | ||
| member _.ToJournal evt = | ||
| match evt with | ||
| | :? Message -> | ||
| box <| Tagged(evt, deft) | ||
| | _ -> evt | ||
|
|
||
|
|
||
| let system = System.create "cluster-system" (configWithPort 0) | ||
| Akka.Cluster.Cluster.Get(system).SelfAddress | ||
| |> Akka.Cluster.Cluster.Get(system).Join | ||
|
|
||
| System.Threading.Thread.Sleep(2000) | ||
|
|
||
| SqlitePersistence.Get(system) |> ignore | ||
|
|
||
| let readJournal = PersistenceQuery.Get(system).ReadJournalFor<SqlReadJournal>(SqlReadJournal.Identifier); | ||
|
|
||
| let source = readJournal.EventsByTag("default") | ||
| let mat = ActorMaterializer.Create(system); | ||
| System.Threading.Thread.Sleep(2000) | ||
| source.RunForeach((fun e ->System.Console.WriteLine(e)), mat) |> ignore | ||
|
|
||
| let actorProp (mailbox : Eventsourced<_>)= | ||
| let rec set (state : Order option) = | ||
| actor { | ||
| let! (msg) = mailbox.Receive() | ||
| match msg with | ||
| | Event (OrderPlaced o) when mailbox.IsRecovering () -> | ||
| return! o |> Some |> set | ||
| | Command(PlaceOrder o) -> | ||
| return o |> OrderPlaced |> Event |> Persist | ||
| | Persisted mailbox (Event(OrderPlaced o)) -> | ||
| return! o |> Some |> set | ||
| | _ -> invalidOp "not supported" | ||
| } | ||
| set None | ||
|
|
||
|
|
||
|
|
||
|
|
||
| let orderFactory = | ||
| (AkklingHelpers.entityFactoryFor system "Order" | ||
| <| propsPersist actorProp | ||
| <| None).RefFor AkklingHelpers.DEFAULT_SHARD | ||
|
|
||
| let init () = mat, system | ||
|
|
||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| module AkklingHelpers | ||
|
|
||
| open System | ||
| open Akka.Actor | ||
| open Akka.Configuration | ||
| open Akka.Cluster | ||
| open Akka.Cluster.Tools.Singleton | ||
| open Akka.Cluster.Sharding | ||
| open Akka.Persistence | ||
| open Akka.Persistence.Sqlite | ||
|
|
||
| open Akkling | ||
| open Akkling.Persistence | ||
| open Akkling.Cluster | ||
| open Akkling.Cluster.Sharding | ||
| open Hyperion | ||
| open Akka.Serialization | ||
|
|
||
| [<Literal>] | ||
| let DEFAULT_SHARD = "default-shard" | ||
|
|
||
| type internal TypedMessageExtractor<'Envelope, 'Message>(extractor: 'Envelope -> string*string*'Message) = | ||
| interface IMessageExtractor with | ||
| member this.ShardId message = | ||
| match message with | ||
| | :? 'Envelope as env -> | ||
| let shardId, _, _ = (extractor(env)) | ||
| shardId | ||
| | :? Akka.Cluster.Sharding.ShardRegion.StartEntity as se -> printfn "%A" se.EntityId; DEFAULT_SHARD | ||
| | _ -> invalidOp <| message.ToString() | ||
| member this.EntityId message = | ||
| match message with | ||
| | :? 'Envelope as env -> | ||
| let _, entityId, _ = (extractor(env)) | ||
| entityId | ||
| | _ ->printfn "kkj"; "entity-1" | ||
| member this.EntityMessage message = | ||
| match message with | ||
| | :? 'Envelope as env -> | ||
| let _, _, msg = (extractor(env)) | ||
| box msg | ||
| | _ -> null | ||
|
|
||
|
|
||
| // HACK over persistent actors | ||
| type FunPersistentShardingActor<'Message>(actor : Eventsourced<'Message> -> Effect<'Message>) as this = | ||
| inherit FunPersistentActor<'Message>(actor) | ||
| // sharded actors are produced in path like /user/{name}/{shardId}/{entityId}, therefore "{name}/{shardId}/{entityId}" is peristenceId of an actor | ||
| let pid = this.Self.Path.Parent.Parent.Name + "/" + this.Self.Path.Parent.Name + "/" + this.Self.Path.Name | ||
| override this.PersistenceId = pid | ||
|
|
||
| // this function hacks persistent functional actors props by replacing them with dedicated sharded version using different PeristenceId strategy | ||
| let internal adjustPersistentProps (props: Props<'Message>) : Props<'Message> = | ||
| if props.ActorType = typeof<FunPersistentActor<'Message>> | ||
| then { props with ActorType = typeof<FunPersistentShardingActor<'Message>> } | ||
| else props | ||
| let entityFactoryFor (system: ActorSystem) (name: string) (props: Props<'Message>) (shardSettings:ClusterShardingSettings option) : EntityFac<'Message> = | ||
|
|
||
| let clusterSharding = ClusterSharding.Get(system) | ||
| let adjustedProps = adjustPersistentProps props | ||
| let shardRegion = | ||
| clusterSharding.Start(name, adjustedProps.ToProps(), | ||
| defaultArg shardSettings (ClusterShardingSettings.Create(system)), new TypedMessageExtractor<_,_>(EntityRefs.entityRefExtractor)) | ||
| { ShardRegion = shardRegion; TypeName = name } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.