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
|
package EventBus
import (
"errors"
"fmt"
"net"
"net/http"
"net/rpc"
"sync"
)
// NetworkBus - object capable of subscribing to remote event buses in addition to remote event
// busses subscribing to it's local event bus. Compoed of a server and client
type NetworkBus struct {
*Client
*Server
service *NetworkBusService
sharedBus Bus
address string
path string
}
// NewNetworkBus - returns a new network bus object at the server address and path
func NewNetworkBus(address, path string) *NetworkBus {
bus := new(NetworkBus)
bus.sharedBus = New()
bus.Server = NewServer(address, path, bus.sharedBus)
bus.Client = NewClient(address, path, bus.sharedBus)
bus.service = &NetworkBusService{&sync.WaitGroup{}, false}
bus.address = address
bus.path = path
return bus
}
// EventBus - returns wrapped event bus
func (networkBus *NetworkBus) EventBus() Bus {
return networkBus.sharedBus
}
// NetworkBusService - object capable of serving the network bus
type NetworkBusService struct {
wg *sync.WaitGroup
started bool
}
// Start - helper method to serve a network bus service
func (networkBus *NetworkBus) Start() error {
var err error
service := networkBus.service
clientService := networkBus.Client.service
serverService := networkBus.Server.service
if !service.started {
server := rpc.NewServer()
server.RegisterName("ServerService", serverService)
server.RegisterName("ClientService", clientService)
server.HandleHTTP(networkBus.path, "/debug"+networkBus.path)
l, e := net.Listen("tcp", networkBus.address)
if e != nil {
err = fmt.Errorf("listen error: %v", e)
}
service.wg.Add(1)
go http.Serve(l, nil)
} else {
err = errors.New("Server bus already started")
}
return err
}
// Stop - signal for the service to stop serving
func (networkBus *NetworkBus) Stop() {
service := networkBus.service
if service.started {
service.wg.Done()
service.started = false
}
}
|
