forked from oklahomer/protoactor-go-sender-example
-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.go
94 lines (78 loc) · 2.38 KB
/
main.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
package main
import (
"github.com/asynkron/protoactor-go/actor"
"github.com/asynkron/protoactor-go/cluster"
"github.com/asynkron/protoactor-go/cluster/clusterproviders/consul"
"github.com/asynkron/protoactor-go/cluster/identitylookup/disthash"
"github.com/asynkron/protoactor-go/remote"
"log"
"os"
"os/signal"
"protoactor-go-sender-example/cluster/messages"
"time"
)
var cnt uint64 = 0
type pingActor struct {
system *actor.ActorSystem
cnt uint
}
func (p *pingActor) Receive(ctx actor.Context) {
switch ctx.Message().(type) {
case struct{}:
cnt += 1
ping := &messages.PingMessage{
Cnt: cnt,
}
grainPid := cluster.GetCluster(p.system).Get("ponger-1", "Ponger")
// Below do not set ctx.Self() as the sender, and hence the recipient has no knowledge of the sender
// even though the message is sent from one actor to another.
//
ctx.Send(grainPid, ping)
case *messages.PongMessage:
// Never comes here because the recipient can not refer to the sender.
// Instead, the cluster grain leaves logs as below:
// YYYY/MM/DD hh:mm:ss Received ping message
// YYYY/MM/DD hh:mm:ss DEBUG [ACTOR] [DeadLetter] pid="<nil>" msg=*messages.PongMessage sender="<nil>"
//
log.Print("Received pong message")
}
}
func main() {
// Set up actor system
system := actor.NewActorSystem()
// Prepare a remote env that listens to 8081
remoteConfig := remote.Configure("127.0.0.1", 8081)
// Configure a cluster on top of the above remote env
cp, err := consul.New()
if err != nil {
log.Fatal(err)
}
lookup := disthash.New()
clusterConfig := cluster.Configure("cluster-example", cp, lookup, remoteConfig)
c := cluster.New(system, clusterConfig)
// Manage the cluster client's lifecycle
c.StartClient() // Configure as a client
defer c.Shutdown(false)
// Start a ping actor that periodically sends a "ping" payload to the "Ponger" cluster grain
pingProps := actor.PropsFromProducer(func() actor.Actor {
return &pingActor{
system: system,
}
})
pingPid := system.Root.Spawn(pingProps)
// Subscribe to a signal to finish the interaction
finish := make(chan os.Signal, 1)
signal.Notify(finish, os.Interrupt, os.Kill)
// Periodically send a ping payload till a signal comes
ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
system.Root.Send(pingPid, struct{}{})
case <-finish:
log.Print("Finish")
return
}
}
}