Module: Fluent::Plugin::FestivalProxy

Included in:
FestivalInput
Defined in:
lib/fluent/plugin/festival_proxy.rb

Defined Under Namespace

Classes: FestivalProxyError

Instance Method Summary collapse

Instance Method Details

#add_location(result, resource) ⇒ Object 



134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 134

def add_location(result, resource)
  if resource.require_location
    log.debug "get_data (location): request #{get_data_request(resource_path(resource.path))}, #{get_data_header.inspect}"
    get_sensor_res = @https.get(get_data_request(resource_path(resource.path)), get_data_header)
    return result if !error_handler(get_sensor_res, "get_data failed.")
    log.debug "get_data: #{get_sensor_res.body}"
    sensor = JSON.parse(get_sensor_res.body)
    # TODO: arbitrary geographicArea type should be supported
    return result.merge({
      "location": {
        "lon": JSON.parse(sensor["location"]["geographicArea"])["coordinates"][0],
        "lat": JSON.parse(sensor["location"]["geographicArea"])["coordinates"][1]
      }
    })
  elsif !resource.fixed_location.nil?
    log.debug "set fixed_location: #{resource.fixed_location.inspect}"
    return result.merge({
      "location": {
        "lon": resource.fixed_location[0].to_f,
        "lat": resource.fixed_location[1].to_f
      }
    })
  end
  return result
end

#create_session ⇒ Object 



56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 56

def create_session
  return @session if valid_session?
  @session_req ||= create_session_request
  session_res = @https.request(@session_req)
  return nil if !error_handler(session_res, 'create_session failed.')
  # access_token is returned as follows
  # {"access_token":"XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX","expires_in":86400}
  @session = JSON.parse(session_res.body)
  # the expiration time is set to 5 minutes before expiration
  @session_expires_in = Time.now + @session["expires_in"] - 5 * 60
end

#create_session_request ⇒ Object 



49
50
51
52
53
54
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 49

def create_session_request
  session_req = Net::HTTP::Post.new(@uri + '/festival/eaas/security/token')
  session_req.body = {email: @email, password: @password}.to_json
  session_req.content_type = 'application/json'
  session_req
end

#delete_session ⇒ Object

TODO: to be implemented



74
75
76
77
78
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 74

def delete_session
  #return if !valid_session?
  #del_session_res = @https.request(delete_session_request)
  #error_handler(del_session_res, 'delete_session failed.')
end

#delete_session_request ⇒ Object

TODO: to be implemented



69
70
71
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 69

def delete_session_request
  #Net::HTTP::Delete.new(@uri + "/session/#{@session_key}")
end

#error_handler(response, message) ⇒ Object 



28
29
30
31
32
33
34
35
36
37
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 28

def error_handler(response, message)
  if response.code != "200"
    log.error error: message
    log.debug "code: #{response.code}"
    log.debug "message: #{response.message}"
    log.debug "body: #{response.body}"
    return false
  end
  return true
end

#get_data ⇒ Object 



160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 160

def get_data
  if !valid_session? && @api_type == 'festival'
    return nil if create_session.nil?
    log.debug "session #{@session} created."
  end
  data = []
  #require 'pry-byebug'
  log.debug "@resources: #{@resources.inspect}"
  @resources.each do |resource|
    case resource_type(resource.path)
    when "current_data" then
      log.debug "get_data: request #{get_data_request(resource.path)}, #{get_data_header.inspect}"
      get_data_res = @https.get(get_data_request(resource.path), get_data_header)
      next if !error_handler(get_data_res,"get_data failed.")
      log.debug "get_data: #{get_data_res.body}"
      result = 
        case @api_type
        when 'festival'
          {
            "resourceName": resource.path,
            "dataValue": JSON.parse(get_data_res.body)["dataValue"]
          }
        when 'sensinact'
          {
            "resourceName": resource.path,
            "dataValue": JSON.parse(get_data_res.body)["response"]["value"],
            "timestamp": (Time.at(JSON.parse(get_data_res.body)["response"]["timestamp"].to_f/1000, JSON.parse(get_data_res.body)["response"]["timestamp"].to_f%1000*1000) if @use_sensor_time)
          }.reject {|k, v| v.nil?}
        else
          return nil
        end
      data << add_location(result, resource)
    when "historical_data" then
      log.error "historical_data is not supported yet"
      next
    else
      log.error "The other resource type is not supported yet"
      log.error "resource_type: #{resource_type(resource.path)}"
      next
    end
  end
  if data.size > 1
    return data
  end
  data[0]
end

#get_data_header ⇒ Object 



106
107
108
109
110
111
112
113
114
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 106

def get_data_header
  header = {
    "Accept": "application/json"
  }
  if @api_type == 'festival'
    return header.merge("X-Auth-Token": @session["access_token"])
  end
  header
end

#get_data_request(path) ⇒ Object

type: “current_data”, “historical_data” def target_path(type)

if !@aggregator_id.nil? && !@testbed_id.nil? && !@resource_id.nil?
  return "/festival/eaas/experimentation/aggregators/#{@aggregator_id}/testbeds/#{@testbed_id}/resources/#{@resource_id}/#{type}"
else
  raise Fluent::ConfigError, "aggregator_id, testbed_id and resource_id must be specified."
end

end



89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 89

def get_data_request(path)
  #get_data_req = @uri + target_path(type)

  get_data_req = @uri +
    case @api_type
    when 'sensinact' 
      Pathname("/sensinact/providers/#{WEBrick::HTTPUtils.escape(path)}").cleanpath.to_s
    else # default (festival)
      Pathname("/festival/eaas/experimentation/#{WEBrick::HTTPUtils.escape(path)}").cleanpath.to_s
    end
  #get_data_req.query = URI.encode_www_form(get_data_params)
  log.debug "#{get_data_req}"
  # currently time window is automatically updated
  #@from = Time.now.iso8601
  get_data_req
end

#get_historical_data ⇒ Object

curl –request GET \

--url 'http://sensinact-cea.ddns.net:8099/festival/driver/testbeds/jose/resources/hyogo001_barometer-info-valueasfloat/historical_data?startDate=2017-03-01T00%3A05%3A55Z' \
--header 'cache-control: no-cache' \
--header 'content-type: application/json' \
--header 'postman-token: 6bceac9e-5d14-3c9e-d34c-acbd4922ebfc' \
--header 'userid: me' \
--data '{"options":{"rows":20}}'



214
215
216
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 214

def get_historical_data
  # TODO: to be implemented
end

#resource_path(path) ⇒ Object 



125
126
127
128
129
130
131
132
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 125

def resource_path(path)
  case @api_type
  when 'sensinact'
    path
  else
    Pathname(path).dirname.to_s
  end
end

#resource_type(path) ⇒ Object 



116
117
118
119
120
121
122
123
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 116

def resource_type(path)
  case @api_type
  when 'sensinact'
    "current_data"
  else
    Pathname(path).basename.to_s
  end
end

#shutdown_proxy ⇒ Object 



22
23
24
25
26
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 22

def shutdown_proxy
  log.debug "shutdown_proxy #{@session.inspect}"
  delete_session
  @https.finish() if @https.active?
end

#start_proxy ⇒ Object 



13
14
15
16
17
18
19
20
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 13

def start_proxy
  log.debug "start festival proxy #{@api_uri}"

  @uri = URI.parse(@api_uri)
  @https = Net::HTTP.new(@uri.host, @uri.port)
  @https.use_ssl = (@uri.scheme == 'https')
  @session = nil
end

#valid_session? ⇒ Boolean

Returns:

  • (Boolean) 


39
40
41
42
43
44
45
46
47
 
# File 'lib/fluent/plugin/festival_proxy.rb', line 39

def valid_session?
  # TODO validate @session by FESTIVAL EaaS API
  if !@session.nil?
    if Time.now < @session_expires_in
      return true
    end
  end
  return false
end