Websocketc: WIP. Trying to get it working with ipcd.
parent
a74bed308b
commit
772389ab41
|
|
@ -22,6 +22,8 @@ targets:
|
|||
main: src/websocketd-dev.cr
|
||||
websocketc:
|
||||
main: src/websocketc.cr
|
||||
oldwsc:
|
||||
main: src/old-websocketc.cr
|
||||
|
||||
# tests
|
||||
test-ws-pong:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,31 @@
|
|||
require "ipc"
|
||||
require "json"
|
||||
require "ipc/json"
|
||||
|
||||
class IPC::JSON
|
||||
def handle(service : IPC::Server, event : IPC::Event::Events)
|
||||
raise "unimplemented"
|
||||
end
|
||||
end
|
||||
|
||||
module Websocketc
|
||||
class_getter requests = [] of IPC::JSON.class
|
||||
class_getter responses = [] of IPC::JSON.class
|
||||
end
|
||||
|
||||
|
||||
class Websocketc::Response
|
||||
IPC::JSON.message Error, 0 do
|
||||
property reason : String | Array(String)
|
||||
|
||||
def initialize(@reason)
|
||||
end
|
||||
end
|
||||
IPC::JSON.message Success, 1 do
|
||||
def initialize
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
require "./requests/*"
|
||||
|
||||
|
|
@ -0,0 +1,83 @@
|
|||
require "http/web_socket"
|
||||
require "option_parser"
|
||||
|
||||
require "ipc"
|
||||
require "./colors"
|
||||
require "./utils"
|
||||
require "./lib_modifications.cr"
|
||||
|
||||
class CLI
|
||||
class_property uri = "ws://localhost:1234/pong"
|
||||
class_property rounds = 1
|
||||
end
|
||||
|
||||
OptionParser.parse do |parser|
|
||||
parser.on "-u uri", "--uri uri", "URI" do |opturi|
|
||||
CLI.uri = opturi
|
||||
end
|
||||
|
||||
parser.on "-r rounds", "--rounds nb-messages", "Nb messages to send." do |opt|
|
||||
CLI.rounds = opt.to_i
|
||||
end
|
||||
|
||||
parser.on "-h", "--help", "Show this help" do
|
||||
puts parser
|
||||
exit 0
|
||||
end
|
||||
end
|
||||
|
||||
def read_then_print(ws : WebSocket)
|
||||
m = read ws
|
||||
puts "message: #{String.new(m)}"
|
||||
end
|
||||
|
||||
def read_then_print_hexa(ws : WebSocket)
|
||||
m = read ws
|
||||
print_hexa(String.new(m), "#{CBLUE}Received message hexa#{CRESET}")
|
||||
end
|
||||
|
||||
def read(ws : WebSocket)
|
||||
puts "reading a message"
|
||||
m = ws.run_once
|
||||
if m.nil?
|
||||
raise "empty message"
|
||||
end
|
||||
|
||||
m
|
||||
end
|
||||
|
||||
def send_with_announce(ws : WebSocket, m : String)
|
||||
puts "sending #{m}"
|
||||
send ws, m
|
||||
end
|
||||
|
||||
def send(ws : WebSocket, m : String | Slice)
|
||||
ws.send m
|
||||
end
|
||||
|
||||
begin
|
||||
ws = WebSocket.new(URI.parse(CLI.uri))
|
||||
puts "connection done: sending pong"
|
||||
|
||||
ws.on_close do |socket|
|
||||
puts "socket is closing"
|
||||
exit 0
|
||||
end
|
||||
|
||||
message = if CLI.uri.ends_with? ".JSON"
|
||||
IPC::Message.new(0, 2.to_u8, 3.to_u8, STDIN.gets_to_end).to_json
|
||||
else
|
||||
IPC::Message.new(0, 2.to_u8, 3.to_u8, STDIN.gets_to_end).to_packet
|
||||
end
|
||||
|
||||
puts "final message: #{message}"
|
||||
CLI.rounds.times do |i|
|
||||
send ws, message
|
||||
pp! read ws
|
||||
end
|
||||
|
||||
ws.close
|
||||
# pp! ws.run_once
|
||||
rescue e
|
||||
puts "Exception: #{e}"
|
||||
end
|
||||
|
|
@ -0,0 +1,81 @@
|
|||
|
||||
class Websocketc::Request
|
||||
IPC::JSON.message Test, 2 do
|
||||
def initialize
|
||||
end
|
||||
|
||||
# def read_then_print(ws : WebSocket)
|
||||
# m = read ws
|
||||
# puts "message: #{String.new(m)}"
|
||||
# end
|
||||
#
|
||||
# def read_then_print_hexa(ws : WebSocket)
|
||||
# m = read ws
|
||||
# print_hexa(String.new(m), "#{CBLUE}Received message hexa#{CRESET}")
|
||||
# end
|
||||
#
|
||||
# def read(ws : WebSocket)
|
||||
# puts "reading a message"
|
||||
# m = ws.run_once
|
||||
# if m.nil?
|
||||
# raise "empty message"
|
||||
# end
|
||||
#
|
||||
# m
|
||||
# end
|
||||
#
|
||||
# def send_with_announce(ws : WebSocket, m : String)
|
||||
# puts "sending #{m}"
|
||||
# send ws, m
|
||||
# end
|
||||
#
|
||||
# def send(ws : WebSocket, m : String | Slice)
|
||||
# ws.send m
|
||||
# end
|
||||
|
||||
|
||||
def handle(websocketc : Websocketc::Service, event : IPC::Event::Events)
|
||||
Baguette::Log.warning "within the test class"
|
||||
|
||||
# ws = WebSocket.new(URI.parse(Websocketc::Context.uri))
|
||||
# puts "connection done: sending pong"
|
||||
#
|
||||
# ws.on_close do |socket|
|
||||
# puts "socket is closing"
|
||||
# exit 0
|
||||
# end
|
||||
#
|
||||
# message = if Websocketc::Context.uri.ends_with? ".JSON"
|
||||
# IPC::Message.new(0, 2.to_u8, 3.to_u8, STDIN.gets_to_end).to_json
|
||||
# else
|
||||
# IPC::Message.new(0, 2.to_u8, 3.to_u8, STDIN.gets_to_end).to_packet
|
||||
# end
|
||||
#
|
||||
# puts "final message: #{message}"
|
||||
# Websocketc::Context.rounds.times do |i|
|
||||
# send ws, message
|
||||
# pp! read ws
|
||||
# end
|
||||
#
|
||||
# ws.close
|
||||
## pp! ws.run_once
|
||||
|
||||
# At the end of this function, we should get something like this.
|
||||
# connections[event.fd] = ws
|
||||
|
||||
Response::Success.new
|
||||
rescue e
|
||||
Baguette::Log.error "during ws init #{e}"
|
||||
Response::Error.new "not implemented"
|
||||
end
|
||||
end
|
||||
Websocketc.requests << Test
|
||||
end
|
||||
|
||||
class Websocketc::Response
|
||||
#IPC::JSON.message TestResponse, 2 do
|
||||
# def initialize
|
||||
# end
|
||||
#end
|
||||
end
|
||||
|
||||
|
|
@ -2,82 +2,174 @@ require "http/web_socket"
|
|||
require "option_parser"
|
||||
|
||||
require "ipc"
|
||||
require "./colors"
|
||||
require "baguette-crystal-base"
|
||||
require "./utils"
|
||||
require "./lib_modifications.cr"
|
||||
|
||||
class CLI
|
||||
class_property uri = "ws://localhost:1234/pong"
|
||||
class_property rounds = 1
|
||||
end
|
||||
|
||||
OptionParser.parse do |parser|
|
||||
parser.on "-u uri", "--uri uri", "URI" do |opturi|
|
||||
CLI.uri = opturi
|
||||
# Websocketc
|
||||
# - is the application allowing libipc communications through Web Socket
|
||||
# - is designed to connect to a Websocketd server, ask for a service, then exchange data
|
||||
# - is WIP
|
||||
|
||||
# Currently, the exchange format is JSON, encapsulated in JSON:
|
||||
# type: Int32 # type of the message
|
||||
# payload: JSON::Any? # encapsulated payload
|
||||
#
|
||||
# Websocketd deserialize the type and the payload before sending it to the service.
|
||||
|
||||
|
||||
module Websocketc
|
||||
class Exception < ::Exception
|
||||
end
|
||||
|
||||
parser.on "-r rounds", "--rounds nb-messages", "Nb messages to send." do |opt|
|
||||
CLI.rounds = opt.to_i
|
||||
end
|
||||
|
||||
parser.on "-h", "--help", "Show this help" do
|
||||
puts parser
|
||||
exit 0
|
||||
class Context
|
||||
class_property uri = "ws://localhost:1234/pong"
|
||||
class_property rounds = 1
|
||||
end
|
||||
end
|
||||
|
||||
def read_then_print(ws : WebSocket)
|
||||
m = read ws
|
||||
puts "message: #{String.new(m)}"
|
||||
class Baguette::Configuration
|
||||
class Websocketc < Base
|
||||
property verbosity : Int32 = 2
|
||||
property print_ipc_timer : Bool = false
|
||||
property ipc_timer : Int32 = 30_000
|
||||
|
||||
def initialize
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def read_then_print_hexa(ws : WebSocket)
|
||||
m = read ws
|
||||
print_hexa(String.new(m), "#{CBLUE}Received message hexa#{CRESET}")
|
||||
end
|
||||
|
||||
def read(ws : WebSocket)
|
||||
puts "reading a message"
|
||||
m = ws.run_once
|
||||
if m.nil?
|
||||
raise "empty message"
|
||||
require "./network.cr"
|
||||
|
||||
|
||||
class Websocketc::Service < IPC::Server
|
||||
property connections : Hash(Int32, WebSocket) = {} of Int32 => WebSocket
|
||||
getter all_fd : Array(Int32) = Array(Int32).new
|
||||
|
||||
property config : Baguette::Configuration::Websocketc
|
||||
|
||||
def initialize(@config : Baguette::Configuration::Websocketc)
|
||||
super "websocketc"
|
||||
end
|
||||
|
||||
m
|
||||
end
|
||||
def handle_request(event : IPC::Event::MessageReceived)
|
||||
request_start = Time.utc
|
||||
|
||||
def send_with_announce(ws : WebSocket, m : String)
|
||||
puts "sending #{m}"
|
||||
send ws, m
|
||||
end
|
||||
# TODO: The message can come from an already linked fd.
|
||||
|
||||
def send(ws : WebSocket, m : String | Slice)
|
||||
ws.send m
|
||||
end
|
||||
request = Websocketc.requests.parse_ipc_json event.message
|
||||
|
||||
begin
|
||||
ws = WebSocket.new(URI.parse(CLI.uri))
|
||||
puts "connection done: sending pong"
|
||||
if request.nil?
|
||||
raise "unknown request type"
|
||||
end
|
||||
|
||||
ws.on_close do |socket|
|
||||
puts "socket is closing"
|
||||
exit 0
|
||||
request_name = request.class.name.sub /^Websocketc::Request::/, ""
|
||||
Baguette::Log.debug "<< #{request_name}"
|
||||
|
||||
response = begin
|
||||
request.handle self, event
|
||||
rescue e : Websocketc::Exception
|
||||
Baguette::Log.error "Websocketc::Exception: #{request_name} => #{e}"
|
||||
Websocketc::Response::Error.new "generic error"
|
||||
rescue e
|
||||
Baguette::Log.error "#{request_name} generic error #{e}"
|
||||
Websocketc::Response::Error.new "unknown error"
|
||||
end
|
||||
|
||||
# If clients sent requests with an “id” field, it is copied
|
||||
# in the responses. Allows identifying responses easily.
|
||||
response.id = request.id
|
||||
|
||||
send event.fd, response
|
||||
|
||||
duration = Time.utc - request_start
|
||||
|
||||
response_name = response.class.name.sub /^Websocketc::Response::/, ""
|
||||
|
||||
if response.is_a? Websocketc::Response::Error
|
||||
Baguette::Log.warning ">> #{response_name} (#{response.reason})"
|
||||
else
|
||||
Baguette::Log.debug ">> #{response_name} (Total duration: #{duration})"
|
||||
end
|
||||
end
|
||||
|
||||
message = if CLI.uri.ends_with? ".JSON"
|
||||
IPC::Message.new(0, 2.to_u8, 3.to_u8, STDIN.gets_to_end).to_json
|
||||
else
|
||||
IPC::Message.new(0, 2.to_u8, 3.to_u8, STDIN.gets_to_end).to_packet
|
||||
def run
|
||||
Baguette::Log.title "Starting websocketc"
|
||||
|
||||
self.loop do |event|
|
||||
begin
|
||||
case event
|
||||
when IPC::Event::Timer
|
||||
Baguette::Log.debug "Timer" if @config.print_ipc_timer
|
||||
|
||||
when IPC::Event::Connection
|
||||
Baguette::Log.debug "connection from #{event.fd}"
|
||||
@all_fd << event.fd
|
||||
|
||||
when IPC::Event::Disconnection
|
||||
Baguette::Log.debug "disconnection from #{event.fd}"
|
||||
@all_fd.select! &.!=(event.fd)
|
||||
connections.delete event.fd
|
||||
|
||||
when IPC::Event::MessageSent
|
||||
Baguette::Log.debug "message sent to #{event.fd}"
|
||||
|
||||
when IPC::Event::MessageReceived
|
||||
Baguette::Log.debug "message received from #{event.fd}"
|
||||
handle_request event
|
||||
|
||||
when IPC::Exception
|
||||
Baguette::Log.warning "IPC::Exception: #{event.message}"
|
||||
|
||||
else
|
||||
Baguette::Log.warning "unhandled IPC event: #{event.class}"
|
||||
end
|
||||
rescue e
|
||||
Baguette::Log.error "exception: #{typeof(e)} - #{e.message}"
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
puts "final message: #{message}"
|
||||
CLI.rounds.times do |i|
|
||||
send ws, message
|
||||
pp! read ws
|
||||
end
|
||||
def self.from_cli
|
||||
|
||||
ws.close
|
||||
# pp! ws.run_once
|
||||
rescue e
|
||||
puts "Exception: #{e}"
|
||||
# First option parsing.
|
||||
simulation, no_configuration, configuration_file = Baguette::Configuration.option_parser
|
||||
|
||||
# Websocketc configuration.
|
||||
configuration = if no_configuration
|
||||
Baguette::Log.info "do not load a configuration file."
|
||||
Baguette::Configuration::Websocketc.new
|
||||
else
|
||||
# In case there is a configuration file helping with the parameters.
|
||||
Baguette::Configuration::Websocketc.get(configuration_file) ||
|
||||
Baguette::Configuration::Websocketc.new
|
||||
end
|
||||
|
||||
|
||||
OptionParser.parse do |parser|
|
||||
parser.on "-u uri", "--uri uri", "URI" do |opturi|
|
||||
Websocketc::Context.uri = opturi
|
||||
end
|
||||
|
||||
parser.on "-r rounds", "--rounds nb-messages", "Nb messages to send." do |opt|
|
||||
Websocketc::Context.rounds = opt.to_i
|
||||
end
|
||||
|
||||
parser.on "-h", "--help", "Show this help" do
|
||||
puts parser
|
||||
exit 0
|
||||
end
|
||||
end
|
||||
|
||||
if simulation
|
||||
pp! configuration
|
||||
exit 0
|
||||
end
|
||||
|
||||
::Websocketc::Service.new configuration
|
||||
end
|
||||
end
|
||||
|
||||
Websocketc::Service.from_cli.run
|
||||
|
|
|
|||
Loading…
Reference in New Issue