grpc_assets.lua

----------------------------------------------------------------------
-- Static binary assets for constructing HTTP/2 responses.
-- These templates are based on the HTTP/2 specification and reference
-- examples from a wireshark capture.
-- (Note: These can be encoded once and reused between requests.)
----------------------------------------------------------------------

local assets = {}

-- Return the binary representation of an annotated ascii hex dump.
-- (See usage below for examples.)
local function hex(str)
   str = str:gsub("#[^\n]+", "") -- remove comments
   str = str:gsub("[^%x]", "")   -- hex only
   return str:gsub("%x%x",       -- hex to binary
                   function(hex) return string.char(tonumber(hex, 16)) end)
end

----------------------------------------------------------------------
-- HEADERS frame for 200 OK 
----------------------------------------------------------------------

-- First a static part.
assets.ok1 = hex [[
  00 00 20       # Length = 32
  01             # Type = Headers
  04             # Flags = End Headers
]]

-- Second the stream ID.

-- Third (last) a static part.
assets.ok2 = hex [[
  # status = 200 OK (static header)
  88
  # content-type: application/grpc (literal header)
  400c636f6e74656e742d74797065106170706c69636174696f6e2f67727063
]]

----------------------------------------------------------------------
-- HEADERS for trailer
----------------------------------------------------------------------

-- First a static part.
assets.trailer1 = hex [[
  00 00 20       # Length = 32
  01             # Type = Headers
  05             # Flags = End Headers + End Stream
]]

-- Second the stream ID.

-- Third (last) a static part.
assets.trailer2 = hex [[
  # grpc-status: 0 (literal header)
  40 0b 67 72 70 63 2d 73 74 61 74 75 73 01 30 
  # grpc-message: OK (literal header)
  40 0c 67 72 70 63 2d 6d 65 73 73 61 67 65 02 4f 4b
]]

----------------------------------------------------------------------
-- DATA frame.
----------------------------------------------------------------------

-- First the length.

-- Second a static part.
assets.data1 = hex [[
  00             # Type = Data
  00             # Flags = (none)
]]

-- Third is the stream ID.

-- Fourth (last) the message payload.

----------------------------------------------------------------------
-- Dynamic parts.
----------------------------------------------------------------------

-- Return a string representing <n> as a <bytes>-long big endian value.
local function bigendian(n, bytes)
  if bytes == 0 then
    return ''
  else
    return bigendian(bit.rshift(n,8), bytes-1)..(string.format('%02x', (n%256)))
  end
end

local function be(n, bytes)
   return (hex(bigendian(n, bytes)))
end

function assets.stream_id(id)
  return be(id, 4)
end

function assets.length(len)
  return be(len, 3)
end

print("stream id 769 = " .. (bigendian(769, 4)) .. " = " .. assets.stream_id(769))

assets.be = be

return assets

grpc_server.lua

-- Fast and minimal HTTP/2 + gRPC server endpoint.
--
-- For use behind an nginx 'grpc_pass' proxy.

local ffi = require("ffi")
local pb = require("pb")
local protoc = require("protoc")
local serpent = require("serpent")

local assets = require("grpc_assets")

local http2_frame_t = ffi.typeof[[
struct {
  uint8_t length[3];
  uint8_t type;
  uint8_t flags;
  int32_t id;
} __attribute__((packed))
]]

local function frame_payload_length(f)
  return f.length[0]*2^16 + f.length[1]*2^8 + f.length[2]
end

local function frame_id(f)
  return bit.bswap(f.id)
end

local http2_frame_ptr_t = ffi.typeof("$*", http2_frame_t)

local http2_payload_t = ffi.typeof[[
union {
  struct { 
    uint16_t id; 
    uint32_t data;
  } settings;
}
]]

------------------------------------------------------------------------
-- Greeter service module
------------------------------------------------------------------------

local schema = protoc:new()

assert(schema:load [[
syntax = "proto3";
package helloworld;

// The greeting service definition.
service Greeter {
  // Sends a greeting
  rpc SayHello (HelloRequest) returns (HelloReply) {}
}

// The request message containing the user's name.
message HelloRequest {
  string name = 1;
}

// The response message containing the greetings
message HelloReply {
  string message = 1;
}
]])

local function serve_greet(socket, payload)
   local message = pb.decode("helloworld.HelloRequest", payload)
   print("\nRequest:\n"..serpent.block(message))
   local reply = { message = "G'day "..message.name }
   local data = pb.encode("helloworld.HelloReply", reply)
   print("\nReply:\n"..serpent.block(pb.decode("helloworld.HelloReply", data)))
   serve_reply(socket, 1, data)
end

------------------------------------------------------------------------
-- 
------------------------------------------------------------------------

function serve_reply (socket, stream_id, message)
   print("got stream_id: "..tostring(stream_id))
   local data = "\x00"..assets.be(#message, 4)..message
   -- HEADERS frame (200 OK; content-type: application/grpc
   socket:send(assets.ok1)
   socket:send(assets.stream_id(stream_id))
   socket:send(assets.ok2)
   -- DATA frame (reply protobuf)
   socket:send(assets.length(#data))
   socket:send(assets.data1)
   socket:send(assets.stream_id(stream_id))
   socket:send(data)
   -- HEADERS frame (grpc-status: 0; grpc-message: OK)
   socket:send(assets.trailer1)
   socket:send(assets.stream_id(stream_id))
   socket:send(assets.trailer2)
   --socket:close()
   --ngx.sleep(10)
end

local function serve_request(socket)
   while true do
      local frame = ffi.cast(http2_frame_ptr_t,
                             assert(socket:receive(ffi.sizeof(http2_frame_t))))
      local payload = assert(socket:receive(frame_payload_length(frame)))
      if     frame.type == 0 then -- data
         print("\nDATA payload:\n"..payload)
         serve_greet(socket, payload)
         return
      elseif frame.type == 1 then -- headers
      elseif frame.type == 2 then -- priority
      elseif frame.type == 3 then -- rst_stream
      elseif frame.type == 4 then -- settings
      elseif frame.type == 5 then -- push_promise
      elseif frame.type == 6 then -- ping
      elseif frame.type == 7 then -- goaway
      elseif frame.type == 8 then -- window_update
      elseif frame.type == 9 then -- continuation
      end
   end
end

local prefix = 'PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n'

local function accept_and_serve()
  local socket = ngx.req.socket(true)
  assert(prefix == socket:receive(#prefix))
  while true do
     return serve_request(socket)
  end
end

return {
  accept_and_serve = accept_and_serve
}

nginx.conf

daemon off;
worker_processes 1;
error_log stderr debug;

events {
}

http {
     access_log stderr;
     upstream lua {
#         server unix:///tmp/grpc.socket max_fails=1;
         server 0.0.0.0:59002;
     }

    server {
        listen 59001 http2;
        location / {
            grpc_pass grpc://lua;
        }
    }
}

stream {


    log_format basic '$remote_addr [$time_local] '
                 '$protocol $status $bytes_sent $bytes_received '
                 '$session_time';

    access_log stderr basic;
    init_by_lua_block {
      grpc_server = require("grpc_server")
    }
    server {
#        listen unix:///tmp/grpc.socket;
        listen 0.0.0.0:59002;
        content_by_lua_block {
            grpc_server.accept_and_serve()
        }
    }
}

HI,
nice work...

can i use this to send a response in haproxy?
i have a grpc client calling haproxy

-haproxy.cfg

http-request use-service lua.authdeny

-service.lua

local function authdeny(applet)
    local pb = require "pb"
    local assets = require("grpc_assets")
    local protoc = require "protoc"

    local protobuf = require('protobuf')
    local input = { code = "403", message = "denied" }
    local message = pb.encode('api.v1.status', input)
    local data = "\x00"..assets.be(#message, 4)..message

    applet:start_response()
    applet:send(assets.ok1)
    applet:send(assets.stream_id(1))
    applet:send(assets.ok2)
    applet:send(assets.length(#data))
    applet:send(assets.data1)
    applet:send(assets.stream_id(1))
    applet:send(data)
end

core.register_service("authdeny", "http", authdeny)