Skip to content

Commit 8cac6bb

Browse files
akkieakkie
committedJul 22, 2020
Move the subscriptions functionality into a separate project
It's currently not possible to start an actor in a gRPC hosted project, therefore the functionality was moved to a HTTP hosted project. See: dapr/dapr#927
1 parent f551429 commit 8cac6bb

17 files changed

+370
-48
lines changed
 

‎README.md

+7
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ The application consists of three services:
1111
| Service | Description | Language |
1212
| ----------------- | ------------------------------- | ---------- |
1313
| `votes` | Handles the voting | F# |
14+
| `subscriptions` | Handles the subscriptions | F# |
1415
| `notifications` | Allows to subscribe for updates | C# |
1516
| `frontend` | The react frontend application | TypeScript |
1617

@@ -73,6 +74,12 @@ dotnet-grpc add-url -o "Protos/dapr/proto/common/v1/common.proto" -i Protos/ -s
7374
dapr run --app-id votes --app-port 3000 -- dotnet run --project "./votes/votes.fsproj"
7475
```
7576

77+
## Run the subscriptions service
78+
79+
```
80+
dapr run --app-id subscriptions --app-port 3001 -- dotnet run --project "./subscriptions/subscriptions.fsproj"
81+
```
82+
7683
## Run the notifications service
7784

7885
The notification service uses a gRPC instead of a REST API. This must be activated by the `--protocol grpc` argument.

‎dapr-vote.sln

+6
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@ Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "votes", "votes\votes.fsproj
88
EndProject
99
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "proto", "proto\proto.csproj", "{4EFEEE95-3CB7-47D0-9049-DFBCDCFB0289}"
1010
EndProject
11+
Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "subscriptions", "subscriptions\subscriptions.fsproj", "{E77CB921-C0D6-4790-B233-FCF797393089}"
12+
EndProject
1113
Global
1214
GlobalSection(SolutionConfigurationPlatforms) = preSolution
1315
Debug|Any CPU = Debug|Any CPU
@@ -30,6 +32,10 @@ Global
3032
{4EFEEE95-3CB7-47D0-9049-DFBCDCFB0289}.Debug|Any CPU.Build.0 = Debug|Any CPU
3133
{4EFEEE95-3CB7-47D0-9049-DFBCDCFB0289}.Release|Any CPU.ActiveCfg = Release|Any CPU
3234
{4EFEEE95-3CB7-47D0-9049-DFBCDCFB0289}.Release|Any CPU.Build.0 = Release|Any CPU
35+
{E77CB921-C0D6-4790-B233-FCF797393089}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
36+
{E77CB921-C0D6-4790-B233-FCF797393089}.Debug|Any CPU.Build.0 = Debug|Any CPU
37+
{E77CB921-C0D6-4790-B233-FCF797393089}.Release|Any CPU.ActiveCfg = Release|Any CPU
38+
{E77CB921-C0D6-4790-B233-FCF797393089}.Release|Any CPU.Build.0 = Release|Any CPU
3339
EndGlobalSection
3440
GlobalSection(NestedProjects) = preSolution
3541
EndGlobalSection

‎vote.http ‎requests.http

+19
Original file line numberDiff line numberDiff line change
@@ -21,3 +21,22 @@ POST http://localhost:3000/votes
2121
Content-Type: application/json
2222

2323
{ "animal": "Cat", "subscription": { "name": "Jon Doe", "email": "jon@doe.com" } }
24+
25+
###
26+
27+
GET http://localhost:3001/subscriptions
28+
Content-Type: application/json
29+
30+
###
31+
32+
POST http://localhost:3001/subscriptions
33+
Content-Type: application/json
34+
35+
{ "name": "Jon Doe", "email": "jon@doe.com" }
36+
37+
###
38+
39+
DELETE http://localhost:3001/subscriptions/jon@doe.com
40+
Content-Type: application/json
41+
42+
###

‎shared/Extensions.fs

+49
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
module Shared.Extensions
2+
3+
open Dapr.Client
4+
open System.Collections.Generic
5+
open Microsoft.FSharp.Control
6+
7+
type DaprClient with
8+
9+
/// <summary>
10+
/// Helper function to handle the non-existence of the state in the data store in a functional way, by returning
11+
/// `None` for null and `Some` for every valid state.
12+
/// </summary>
13+
/// <param name="storeName">The name of state store to read from.</param>
14+
/// <param name="key">The state key.</param>
15+
/// <returns>Some 'T if the key exists in store, None otherwise.</returns>
16+
member this.GetStateAsyncO<'T>(storeName: string, key: string): Async<Option<'T>> =
17+
async {
18+
let! state = this.GetStateAsync<'T>(storeName, key).AsTask() |> Async.AwaitTask
19+
20+
return if isNull (box state) then None else Some(state)
21+
}
22+
23+
/// <summary>
24+
/// Helper function to handle the non-existence of the state in the data store in a functional way, by returning
25+
/// `defaultValue` for null and `Some` for every valid state.
26+
/// </summary>
27+
/// <param name="storeName">The name of state store to read from.</param>
28+
/// <param name="key">The state key.</param>
29+
/// <param name="defaultValue">The value to return if the result was `None`.</param>
30+
/// <returns>'T if the key exists in store, `defaultValue` otherwise.</returns>
31+
member this.GetStateAsyncOr<'T>(storeName: string, key: string, defaultValue: 'T): Async<'T> =
32+
async {
33+
let! state = this.GetStateAsyncO(storeName, key)
34+
35+
return state |> Option.defaultValue(defaultValue)
36+
}
37+
38+
type Dictionary<'A, 'B> with
39+
40+
/// <summary>
41+
/// Helper function to handle the non-existence of the key in the dictionary in a functional way, by returning
42+
/// `None` for null and `Some` for the existing key.
43+
/// </summary>
44+
/// <param name="key">The state key.</param>
45+
/// <returns>Some 'T if the key exists iun store, None otherwise.</returns>
46+
member this.GetValueO(key: 'A): Option<'B> =
47+
let value = this.GetValueOrDefault<'A, 'B>(key)
48+
49+
if isNull (box value) then None else Some(value)

‎shared/Extensions/DaprClient.fs

-19
This file was deleted.

‎votes/Vote.fs ‎shared/Models.fs

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
namespace Votes
1+
namespace Shared
22

33
type Animal =
44
| Cat

‎shared/shared.fsproj

+2-1
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,9 @@
44
</PropertyGroup>
55

66
<ItemGroup>
7-
<Compile Include="Extensions/DaprClient.fs" />
87
<Compile Include="Config.fs" />
8+
<Compile Include="Extensions.fs" />
9+
<Compile Include="Models.fs" />
910
</ItemGroup>
1011

1112
<ItemGroup>

‎subscriptions/Actors/SubscriptionActor.fs

+95
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
namespace Subscriptions.Actors
2+
3+
open Dapr.Client
4+
open Dapr.Actors
5+
open Dapr.Actors.Runtime
6+
open FSharpx.Control
7+
open System.Collections.Generic
8+
open System.Threading.Tasks
9+
open Shared.Config
10+
open Shared.Extensions
11+
open Shared
12+
13+
type Subscriptions = Dictionary<string, Subscription>
14+
15+
module SubscriptionActor =
16+
17+
/// <summary>
18+
/// The name of the actor.
19+
/// </summary>
20+
[<Literal>]
21+
let Name = "SubscriptionActor"
22+
23+
/// <summary>
24+
/// The ID of the actor.
25+
/// </summary>
26+
let ID = ActorId("subscription")
27+
28+
/// <summary>
29+
/// An actor that is responsible for handling the subscriptions in the state store.
30+
/// </summary>
31+
type ISubscriptionActor =
32+
inherit IActor
33+
34+
/// <summary>
35+
/// Adds the given subscriptions to the state store.
36+
/// </summary>
37+
/// <param name="subscription">The subscription data.</param>
38+
/// <returns>A task that represents the asynchronous operation.</returns>
39+
abstract Subscribe: subscription: Subscription -> Task<Subscriptions>
40+
41+
/// <summary>
42+
/// Removes the subscription for the given email from the state store.
43+
/// </summary>
44+
/// <param name="email">The email to unsubscribe.</param>
45+
/// <returns>A task that represents the asynchronous operation.</returns>
46+
abstract Unsubscribe: email: string -> Task<Subscriptions>
47+
48+
/// <summary>
49+
/// Concrete subscription actor implementation.
50+
///
51+
/// Storing all subscriptions under a single key, may be an overhead for a large set of subscriptions. For
52+
/// production usage, an other storage format may be better.
53+
/// </summary>
54+
[<Actor(TypeName = SubscriptionActor.Name)>]
55+
type SubscriptionActor(actorService: ActorService, actorId: ActorId, daprClient: DaprClient) =
56+
inherit Actor(actorService, actorId)
57+
58+
interface ISubscriptionActor with
59+
60+
/// <inheritdoc/>
61+
member _.Subscribe(subscription: Subscription) =
62+
async {
63+
let! subscriptions =
64+
daprClient.GetStateAsyncOr(StateStore.name, StateStore.subscriptions, Subscriptions())
65+
66+
let maybeValue = subscriptions.GetValueO(subscription.Email)
67+
68+
return!
69+
match maybeValue with
70+
| Some(value) when value.Name = subscription.Name ->
71+
async { return subscriptions }
72+
| _ ->
73+
subscriptions.[subscription.Email] <- subscription
74+
75+
daprClient.SaveStateAsync(StateStore.name, StateStore.subscriptions, subscriptions)
76+
|> Async.AwaitTask
77+
|> Async.map (fun _ -> subscriptions)
78+
}
79+
|> Async.StartAsTask
80+
81+
/// <inheritdoc/>
82+
member _.Unsubscribe(email: string) =
83+
async {
84+
let! subscriptions =
85+
daprClient.GetStateAsyncOr(StateStore.name, StateStore.subscriptions, Subscriptions())
86+
87+
return!
88+
match subscriptions.Remove(email) with
89+
| true ->
90+
daprClient.SaveStateAsync(StateStore.name, StateStore.subscriptions, subscriptions)
91+
|> Async.AwaitTask
92+
|> Async.map (fun _ -> subscriptions)
93+
| false -> async { return subscriptions }
94+
}
95+
|> Async.StartAsTask

‎subscriptions/Controllers/SubscriptionController.fs

+42
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
namespace Subscriptions.Controllers
2+
3+
open Dapr.Client
4+
open Dapr.Actors.Client
5+
open Microsoft.AspNetCore.Mvc
6+
open Shared.Config
7+
open Shared.Extensions
8+
open Shared
9+
open Subscriptions.Actors
10+
11+
[<ApiController>]
12+
type VoteController([<FromServices>] daprClient: DaprClient) =
13+
inherit ControllerBase()
14+
15+
[<HttpGet("subscriptions")>]
16+
member _.Subscriptions() =
17+
async {
18+
let! votes =
19+
daprClient.GetStateAsyncOr<Subscriptions>(StateStore.name, StateStore.subscriptions, Subscriptions())
20+
21+
return OkObjectResult(votes)
22+
}
23+
24+
[<HttpPost("subscriptions")>]
25+
member _.Subscribe([<FromBody>] subscription: Subscription) =
26+
async {
27+
let proxy = ActorProxy.Create<ISubscriptionActor>(SubscriptionActor.ID, SubscriptionActor.Name)
28+
29+
proxy.Subscribe(subscription) |> Async.AwaitTask |> ignore
30+
31+
return OkResult()
32+
}
33+
34+
[<HttpDelete("subscriptions/{email}")>]
35+
member _.Unsubscribe(email: string) =
36+
async {
37+
let proxy = ActorProxy.Create<ISubscriptionActor>(SubscriptionActor.ID, SubscriptionActor.Name)
38+
39+
proxy.Unsubscribe(email) |> Async.AwaitTask |> ignore
40+
41+
return OkResult()
42+
}

‎subscriptions/Program.fs

+38
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
namespace Subscriptions
2+
3+
open Dapr.Actors.AspNetCore
4+
open Dapr.Actors.Runtime
5+
open Microsoft.AspNetCore.Hosting
6+
open Microsoft.Extensions.Hosting
7+
open Shared.Config
8+
open Subscriptions.Actors
9+
10+
module Program =
11+
let exitCode = 0
12+
let port = 3001
13+
14+
let CreateHostBuilder args =
15+
Host.CreateDefaultBuilder(args)
16+
.ConfigureWebHostDefaults(fun webBuilder ->
17+
webBuilder
18+
.UseStartup<Startup>()
19+
.UseActors(fun actorRuntime ->
20+
actorRuntime.RegisterActor<SubscriptionActor>(fun tpe ->
21+
ActorService(tpe, fun actorService actorId ->
22+
SubscriptionActor(
23+
actorService,
24+
actorId,
25+
Dapr.client
26+
) :> Actor
27+
)
28+
)
29+
)
30+
.UseUrls(sprintf "http://localhost:%d/" port)
31+
|> ignore
32+
)
33+
34+
[<EntryPoint>]
35+
let main args =
36+
CreateHostBuilder(args).Build().Run()
37+
38+
exitCode

‎subscriptions/Properties/launchSettings.json

+30
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
{
2+
"$schema": "http://json.schemastore.org/launchsettings.json",
3+
"iisSettings": {
4+
"windowsAuthentication": false,
5+
"anonymousAuthentication": true,
6+
"iisExpress": {
7+
"applicationUrl": "http://localhost:57778",
8+
"sslPort": 44326
9+
}
10+
},
11+
"profiles": {
12+
"IIS Express": {
13+
"commandName": "IISExpress",
14+
"launchBrowser": true,
15+
"launchUrl": "vote",
16+
"environmentVariables": {
17+
"ASPNETCORE_ENVIRONMENT": "Development"
18+
}
19+
},
20+
"votes": {
21+
"commandName": "Project",
22+
"launchBrowser": true,
23+
"launchUrl": "vote",
24+
"applicationUrl": "https://localhost:5001;http://localhost:5000",
25+
"environmentVariables": {
26+
"ASPNETCORE_ENVIRONMENT": "Development"
27+
}
28+
}
29+
}
30+
}

‎subscriptions/Startup.fs

+42
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
namespace Subscriptions
2+
3+
open Dapr.Client
4+
open Shared.Config
5+
open Microsoft.AspNetCore.Builder
6+
open Microsoft.AspNetCore.Hosting
7+
open Microsoft.Extensions.Configuration
8+
open Microsoft.Extensions.DependencyInjection
9+
open Microsoft.Extensions.Hosting
10+
11+
type Startup private () =
12+
13+
new(configuration: IConfiguration) as this =
14+
Startup()
15+
then this.Configuration <- configuration
16+
17+
member this.ConfigureServices(services: IServiceCollection) =
18+
services
19+
.AddControllers()
20+
.AddDapr()
21+
.AddJsonOptions(fun options -> options.JsonSerializerOptions.Converters.Add(jsonConverter))
22+
|> ignore
23+
24+
services.AddRouting() |> ignore
25+
services.AddSingleton(jsonSerializerOptions) |> ignore
26+
services.AddSingleton<DaprClient>(Dapr.client) |> ignore
27+
28+
member this.Configure(app: IApplicationBuilder, env: IWebHostEnvironment) =
29+
if (env.IsDevelopment()) then
30+
app.UseDeveloperExceptionPage() |> ignore
31+
else
32+
app.UseHsts() |> ignore
33+
34+
app.UseRouting() |> ignore
35+
app.UseCloudEvents() |> ignore
36+
app.UseAuthorization() |> ignore
37+
app.UseEndpoints(fun endpoints ->
38+
endpoints.MapSubscribeHandler() |> ignore
39+
endpoints.MapControllers() |> ignore)
40+
|> ignore
41+
42+
member val Configuration: IConfiguration = null with get, set

‎subscriptions/subscriptions.fsproj

+21
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
<Project Sdk="Microsoft.NET.Sdk.Web">
2+
<PropertyGroup>
3+
<TargetFramework>netcoreapp3.1</TargetFramework>
4+
</PropertyGroup>
5+
<ItemGroup>
6+
<ProjectReference Include="../shared/shared.fsproj" />
7+
<ProjectReference Include="../proto/proto.csproj" />
8+
</ItemGroup>
9+
<ItemGroup>
10+
<Compile Include="Actors/SubscriptionActor.fs" />
11+
<Compile Include="Controllers/SubscriptionController.fs" />
12+
<Compile Include="Startup.fs" />
13+
<Compile Include="Program.fs" />
14+
</ItemGroup>
15+
<ItemGroup>
16+
<PackageReference Include="Dapr.Actors" Version="0.9.0-preview01" />
17+
<PackageReference Include="Dapr.Actors.AspNetCore" Version="0.9.0-preview01" />
18+
<PackageReference Include="Dapr.AspNetCore" Version="0.9.0-preview01" />
19+
<PackageReference Include="FSharpx.Async" Version="1.14.1" />
20+
</ItemGroup>
21+
</Project>

‎votes/Actors/VotingActor.fs

+8-13
Original file line numberDiff line numberDiff line change
@@ -3,10 +3,11 @@ namespace Votes.Actors
33
open Dapr.Client
44
open Dapr.Actors
55
open Dapr.Actors.Runtime
6-
open System.Threading.Tasks
7-
open Shared.Extensions
6+
open FSharpx.Control
87
open Shared.Config
9-
open Votes
8+
open Shared.Extensions
9+
open System.Threading.Tasks
10+
open Shared
1011

1112
module VotingActor =
1213

@@ -49,18 +50,12 @@ type VotingActor(actorService: ActorService, actorId: ActorId, daprClient: DaprC
4950
/// <inheritdoc/>
5051
member _.Vote(animal: Animal) =
5152
async {
52-
let! maybeVotes = daprClient.GetStateAsyncF<Votes>(StateStore.name, StateStore.votes)
53+
let! votes = daprClient.GetStateAsyncOr(StateStore.name, StateStore.votes, Votes.empty)
5354

54-
let updatedVotes =
55-
(match maybeVotes with
56-
| Some (votes) -> votes
57-
| None -> Votes.empty)
58-
.Vote(animal)
55+
let updatedVotes = votes.Vote(animal)
5956

60-
daprClient.SaveStateAsync<Votes>(StateStore.name, StateStore.votes, updatedVotes)
57+
return! daprClient.SaveStateAsync<Votes>(StateStore.name, StateStore.votes, updatedVotes)
6158
|> Async.AwaitTask
62-
|> ignore
63-
64-
return updatedVotes
59+
|> Async.map(fun _ -> updatedVotes)
6560
}
6661
|> Async.StartAsTask

‎votes/Controllers/VoteController.fs

+9-11
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,11 @@
22

33
open Dapr.Client
44
open Dapr.Actors.Client
5+
open Dapr.Client.Http
56
open Microsoft.AspNetCore.Mvc
6-
open Notifications
77
open Shared.Config
88
open Shared.Extensions
9-
open Votes
9+
open Shared
1010
open Votes.Actors
1111

1212
[<ApiController>]
@@ -16,12 +16,7 @@ type VoteController([<FromServices>] daprClient: DaprClient) =
1616
[<HttpGet("votes")>]
1717
member _.Results() =
1818
async {
19-
let! maybeVotes = daprClient.GetStateAsyncF<Votes>(StateStore.name, StateStore.votes)
20-
21-
let votes =
22-
match maybeVotes with
23-
| Some (votes) -> votes
24-
| None -> Votes.empty
19+
let! votes = daprClient.GetStateAsyncOr<Votes>(StateStore.name, StateStore.votes, Votes.empty)
2520

2621
return OkObjectResult(votes)
2722
}
@@ -34,11 +29,14 @@ type VoteController([<FromServices>] daprClient: DaprClient) =
3429
let! votes = proxy.Vote(vote.Animal) |> Async.AwaitTask
3530

3631
match vote.Subscription with
32+
| None -> ()
3733
| Some(subscription) ->
38-
let grpcSubscription = Grpc.Subscription(Name = subscription.Name, Email = subscription.Email)
39-
daprClient.InvokeMethodAsync<Grpc.Subscription>("notifications", "Subscribe", grpcSubscription)
34+
let httpExtension = HTTPExtension()
35+
httpExtension.Verb <- HTTPVerb.Post
36+
httpExtension.ContentType <- "application/json"
37+
38+
daprClient.InvokeMethodAsync<Subscription>("subscriptions", "subscriptions", subscription, httpExtension)
4039
|> Async.AwaitTask |> ignore
41-
| None -> ()
4240

4341
return OkObjectResult(votes)
4442
}

‎votes/Program.fs

-1
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@ module Program =
1919
.UseActors(fun actorRuntime ->
2020
actorRuntime.RegisterActor<VotingActor>(fun tpe ->
2121
ActorService(tpe, fun actorService actorId ->
22-
printfn "ActorID: %s" (actorId.ToString())
2322
VotingActor(
2423
actorService,
2524
actorId,

‎votes/votes.fsproj

+1-2
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77
<ProjectReference Include="../proto/proto.csproj" />
88
</ItemGroup>
99
<ItemGroup>
10-
<Compile Include="Vote.fs" />
1110
<Compile Include="Actors/VotingActor.fs" />
1211
<Compile Include="Controllers/VoteController.fs" />
1312
<Compile Include="Startup.fs" />
@@ -17,6 +16,6 @@
1716
<PackageReference Include="Dapr.Actors" Version="0.9.0-preview01" />
1817
<PackageReference Include="Dapr.Actors.AspNetCore" Version="0.9.0-preview01" />
1918
<PackageReference Include="Dapr.AspNetCore" Version="0.9.0-preview01" />
20-
<PackageReference Include="FSharp.SystemTextJson" Version="0.11.13" />
19+
<PackageReference Include="FSharpx.Async" Version="1.14.1" />
2120
</ItemGroup>
2221
</Project>

0 commit comments

Comments
 (0)
Please sign in to comment.