Class: RubyPitaya::NatsConnector

Inherits:
Object
  • Object
show all
Defined in:
lib/rubypitaya/core/nats_connector.rb

Constant Summary collapse

REQUEST_TYPE_SYS = 
0
REQUEST_TYPE_USER = 
1
MESSAGE_TYPE_REQUEST = 
0
MESSAGE_TYPE_NOTIFY = 
1
MESSAGE_TYPE_RESPONSE = 
2
MESSAGE_TYPE_PUSH = 
3

Instance Method Summary collapse

Constructor Details

#initialize(nats_address, server_uuid, server_name) ⇒ NatsConnector

Returns a new instance of NatsConnector.



16
17
18
19
20
 
# File 'lib/rubypitaya/core/nats_connector.rb', line 16

def initialize(nats_address, server_uuid, server_name)
  @server_uuid = server_uuid
  @server_name = server_name
  @nats_address = nats_address
end

Instance Method Details

#connect ⇒ Object 



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/rubypitaya/core/nats_connector.rb', line 22

def connect
  NATS.start(:servers => [@nats_address]) do |nc|

    NATS.subscribe(subscribe_topic) do |message, reply, a, b|
      request = NatsRequest.decode(message).to_hash

      Fiber.new do
        response = yield request

        nats_response = NatsResponse.new(data: response.to_json)

        NATS.publish(reply, nats_response)
      end.resume
    end

  end
end

#get_frontend_topic(frontend_id) ⇒ Object 



88
89
90
 
# File 'lib/rubypitaya/core/nats_connector.rb', line 88

def get_frontend_topic(frontend_id)
  "pitaya/servers/connector/#{frontend_id}"
end

#get_user_message_topic(uid) ⇒ Object 



92
93
94
 
# File 'lib/rubypitaya/core/nats_connector.rb', line 92

def get_user_message_topic(uid)
  "pitaya/connector/user/#{uid}/push"
end

#push_to_frontend(session, message_route, payload) ⇒ Object 



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
 
# File 'lib/rubypitaya/core/nats_connector.rb', line 40

def push_to_frontend(session, message_route, payload)
  frontend_topic = get_frontend_topic(session.frontend_id)

  nats_message = NatsMessage.new(
    route: message_route,
    data: payload,
    reply: @server_uuid,
    type: MESSAGE_TYPE_REQUEST,
  )

  nats_session = NatsSession.new(
    id: session.id,
    uid: session.uid,
    data: session.data.to_json,
  )

  nats_request = NatsRequest.new(
    type: REQUEST_TYPE_USER,
    session: nats_session,
    msg: nats_message,
    frontendID: session.frontend_id,
    metadata: session.metadata.to_json,
  )

  nats_response = NATS.request(frontend_topic, nats_request)
  response = NatsResponse.decode(nats_response).to_hash

  response
end

#push_to_user(uid, message_route, payload) ⇒ Object 



70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/rubypitaya/core/nats_connector.rb', line 70

def push_to_user(uid, message_route, payload)
  user_topic = get_user_message_topic(uid)

  nats_push = NatsPush.new(
    route: message_route,
    uid: uid,
    data: payload.to_json,
  )

  NATS.publish(user_topic, nats_push)

  0
end

#subscribe_topic ⇒ Object 



84
85
86
 
# File 'lib/rubypitaya/core/nats_connector.rb', line 84

def subscribe_topic
  "pitaya/servers/#{@server_name}/#{@server_uuid}"
end