n2o_stream.erl 1.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445
  1. -module(n2o_stream).
  2. -description('N4U Stream Bridge to WebSocket or XHR channels').
  3. -behaviour(cowboy_http_handler).
  4. -behaviour(cowboy_websocket_handler).
  5. -export([init/3,handle/2,info/3,terminate/3]).
  6. -export([websocket_init/3,websocket_handle/3,websocket_info/3,websocket_terminate/3]).
  7. % XHR
  8. init(T,R,O) -> upgrade(cowboy_req:header(<<"upgrade">>,R),{T,R,O}).
  9. websocket(_R,<<"websocket">>) -> {upgrade, protocol, cowboy_websocket};
  10. websocket(R,_) -> down(reply([],R,501)).
  11. upgrade({undefined,R},{T,R,O}) -> initialize(T,R,O);
  12. upgrade({B,R},_) when is_binary(B) -> websocket(R,cowboy_bstr:to_lower(B)).
  13. handle(R,S) -> body(cowboy_req:body(R),S).
  14. info(M,R,S) -> xhr(n4u_proto:info(M,R,S)).
  15. terminate(_,R,S) -> n4u_proto:terminate(R,S).
  16. initialize(T,R,O) -> xhr(n4u_proto:init(T,R,[{formatter,json}|O],xhr)).
  17. body({ok,D,R2},S) -> xhr(n4u_proto:stream({type(D),D},R2,S));
  18. body(R,S) -> {ok,R,S}.
  19. down(R) -> {shutdown,R,undefined}.
  20. reply(D,R,Code) -> {ok,R2}=cowboy_req:reply(Code,[],D,R), R2.
  21. type(<<"N4U,",_/binary>>) -> text;
  22. type(<<"PING">>) -> text;
  23. type(_) -> binary.
  24. xhr({ok,R,S}) -> {ok,R,S};
  25. xhr({shutdown,R,S}) -> {shutdown,R,S};
  26. xhr({reply,D,R,S}) -> {ok,reply(D,R,200),S}.
  27. % WebSocket
  28. websocket_info(I,R,S) -> ws(n4u_proto:info(I,R,S)).
  29. websocket_handle(D,R,S) -> ws(n4u_proto:stream(D,R,S)).
  30. %websocket_handle(_,R,S) -> {ok,R,S,hibernate}.
  31. websocket_init(T,R,O) -> ws(n4u_proto:init(T,R,[{formatter,bert}|O],ws)).
  32. websocket_terminate(_,R,S) -> n4u_proto:terminate(R,S).
  33. ws({ok,R,S}) -> {ok,R,S,hibernate};
  34. ws({shutdown,R,S}) -> {shutdown,R,S};
  35. ws({reply,{binary,Rep},R,S}) -> {reply,{binary,Rep},R,S,hibernate};
  36. ws({reply,Rep,R,S}) -> {reply,{text,Rep},R,S,hibernate}.