implement unit stream (#11)

Cargo.toml

@@ -20,6 +20,7 @@ prost = "0.7"
 serde = { version = "1.0", features = ["derive"] }
 thiserror = "1.0"
 tokio = { version = "1.0", features = ["rt-multi-thread", "time", "sync"] }
+tokio-stream = "0.1"
 tonic = "0.4"
 
 [dev-dependencies]

lua/exporters/object.lua

@@ -20,6 +20,15 @@ GRPC.exporters.unit = function(unit)
   }
 end
 
+GRPC.exporters.group = function(group)
+  return {
+    id = tonumber(group:getID()),
+    name = group:getName(),
+    coalition = group:getCoalition(),
+    category = group:getCategory(),
+  }
+end
+
 GRPC.exporters.weapon = function(weapon)
   return {
     id = tonumber(weapon:getName()),

lua/grpc.lua

@@ -78,13 +78,14 @@ end
 --
 
 GRPC.methods = {}
-dofile(GRPC.basePath .. [[methods\coalitions.lua]])
 dofile(GRPC.basePath .. [[methods\atmosphere.lua]])
+dofile(GRPC.basePath .. [[methods\coalitions.lua]])
+dofile(GRPC.basePath .. [[methods\custom.lua]])
+dofile(GRPC.basePath .. [[methods\group.lua]])
+dofile(GRPC.basePath .. [[methods\mission.lua]])
 dofile(GRPC.basePath .. [[methods\trigger.lua]])
 dofile(GRPC.basePath .. [[methods\unit.lua]])
 dofile(GRPC.basePath .. [[methods\world.lua]])
-dofile(GRPC.basePath .. [[methods\custom.lua]])
-dofile(GRPC.basePath .. [[methods\event_stream.lua]])
 
 --
 -- RPC request handler

lua/methods/coalitions.lua

@@ -15,5 +15,20 @@ GRPC.methods.getPlayers = function(params)
     result[i] = GRPC.exporters.unit(unit)
   end
   return GRPC.success({units = result})
+end
 
-end
+GRPC.methods.getGroups = function(params)
+  local result = {}
+  for _, c in pairs(coalition.side) do
+    if params.coalition == nil or params.coalition == c then
+      -- https://wiki.hoggitworld.com/view/DCS_func_getGroups
+      local groups = coalition.getGroups(c, params.category)
+
+      for _, group in ipairs(groups) do
+        table.insert(result, GRPC.exporters.group(group))
+      end
+    end
+  end
+
+  return GRPC.success({groups = result})
+end

lua/methods/group.lua

@@ -0,0 +1,26 @@
+--
+-- RPC unit actions
+-- https://wiki.hoggitworld.com/view/DCS_Class_Group
+--
+
+local GRPC = GRPC
+
+GRPC.methods.getUnits = function(params)
+  -- https://wiki.hoggitworld.com/view/DCS_func_getByName
+  local group = Group.getByName(params.groupName)
+  if group == nil then
+    return GRPC.errorNotFound("group does not exist")
+  end
+
+  -- https://wiki.hoggitworld.com/view/DCS_func_getUnits
+  local units = group:getUnits()
+
+  local result = {}
+  for i, unit in ipairs(units) do
+    if params.active == nil or params.active == unit:isActive() then
+      result[i] = GRPC.exporters.unit(unit)
+    end
+  end
+
+  return GRPC.success({units = result})
+end

lua/methods/event_stream.lua → lua/methods/mission.lua

@@ -261,7 +261,7 @@ GRPC.onDcsEvent = function(event)
       type = "markAdd",
       initiator = exporter(event.initiator),
       id = event.idx,
-      pos = GRPC.toLatLonPosition (event.pos),
+      pos = GRPC.toLatLonPosition(event.pos),
       text = event.text,
     }
     if event.groupID > -1 and event.groupID then
@@ -279,7 +279,7 @@ GRPC.onDcsEvent = function(event)
       type = "markChange",
       initiator = exporter(event.initiator),
       id = event.idx,
-      pos = GRPC.toLatLonPosition (event.pos),
+      pos = GRPC.toLatLonPosition(event.pos),
       text = event.text,
     }
     if event.groupID > -1 and event.groupID then
@@ -297,7 +297,7 @@ GRPC.onDcsEvent = function(event)
       type = "markRemove",
       initiator = exporter(event.initiator),
       id = event.idx,
-      pos = GRPC.toLatLonPosition (event.pos),
+      pos = GRPC.toLatLonPosition(event.pos),
       text = event.text,
     }
     if event.groupID > -1 and event.groupID then

lua/methods/unit.lua

@@ -45,4 +45,30 @@ GRPC.methods.getRadar = function(params)
     active = active,
     target = grpcTable
   })
-end
+end
+
+GRPC.methods.getUnitPosition = function(params)
+  -- https://wiki.hoggitworld.com/view/DCS_func_getByName
+  local unit = Unit.getByName(params.name)
+  if unit == nil then
+    return GRPC.errorNotFound("unit does not exist")
+  end
+
+  return GRPC.success({
+    -- https://wiki.hoggitworld.com/view/DCS_func_getPoint
+    position = GRPC.toLatLonPosition(unit:getPoint()),
+  })
+end
+
+GRPC.methods.getUnitPlayerName = function(params)
+  -- https://wiki.hoggitworld.com/view/DCS_func_getByName
+  local unit = Unit.getByName(params.name)
+  if unit == nil then
+    return GRPC.errorNotFound("unit does not exist")
+  end
+
+  return GRPC.success({
+    -- https://wiki.hoggitworld.com/view/DCS_func_getPlayerName
+    playerName = unit:getPlayerName(),
+  })
+end

protos/coalitions.proto

@@ -1,6 +1,7 @@
 syntax = "proto3";
 
 import "common.proto";
+import "group.proto";
 
 package dcs;
 
@@ -10,4 +11,13 @@ message GetPlayersRequest {
 
 message GetPlayersResponse {
 	repeated Unit units = 1;
-}
+}
+
+message GetGroupsRequest {
+  Coalition coalition = 1;
+  optional dcs.group.GroupCategory category = 2;
+}
+
+message GetGroupsResponse {
+  repeated Group groups = 1;
+}

protos/common.proto

@@ -45,6 +45,11 @@ message Unit {
   optional string player_name = 7;
 }
 
+message Group {
+  uint32 id = 1;
+  string name = 2;
+}
+
 message Weapon {
 	uint32 id = 1;
 	string type = 2;

protos/dcs.proto

@@ -1,13 +1,14 @@
 syntax = "proto3";
 
 import "atmosphere.proto";
-import "event_stream.proto";
+import "coalitions.proto";
+import "common.proto";
+import "custom.proto";
+import "group.proto";
+import "mission.proto";
 import "trigger.proto";
 import "unit.proto";
 import "world.proto";
-import "custom.proto";
-import "common.proto";
-import "coalitions.proto";
 
 package dcs;
 
@@ -25,6 +26,9 @@ service Atmosphere {
 service Coalitions {
 	// https://wiki.hoggitworld.com/view/DCS_func_getPlayers
 	rpc GetPlayers(GetPlayersRequest) returns (GetPlayersResponse) {}
+
+	// https://wiki.hoggitworld.com/view/DCS_func_getGroups
+	rpc GetGroups(GetGroupsRequest) returns (GetGroupsResponse) {}
 }
 
 service Custom {
@@ -41,9 +45,7 @@ service Mission {
 	rpc StreamEvents(StreamEventsRequest) returns (stream Event) {}
 
 	// Streams unit updates
-	// TODO https://github.com/DCS-gRPC/rust-server/issues/8
-	// Envisioned RPC Signature: 
-	// rpc StreamUnits(StreamUnitsRequest) returns (stream Unit) {}
+	rpc StreamUnits(StreamUnitsRequest) returns (stream UnitUpdate) {}
 }
 
 service Triggers {
@@ -87,9 +89,20 @@ service Triggers {
 	rpc SignalFlare(SignalFlareRequest) returns (EmptyResponse) {}
 }
 
+service Groups {
+	// https://wiki.hoggitworld.com/view/DCS_func_getUnits
+	rpc GetUnits(dcs.group.GetUnitsRequest) returns (dcs.group.GetUnitsResponse) {}
+}
+
 service Units {
 	// https://wiki.hoggitworld.com/view/DCS_func_getRadar
-	rpc GetRadar(GetRadarRequest) returns (GetRadarResponse) {}
+	rpc GetRadar(UnitName) returns (GetRadarResponse) {}
+
+	// https://wiki.hoggitworld.com/view/DCS_func_getPoint
+	rpc GetPosition(UnitName) returns (GetUnitPositionResponse) {}
+
+	// https://wiki.hoggitworld.com/view/DCS_func_getPlayerName
+	rpc GetPlayerName(UnitName) returns (GetUnitPlayerNameResponse) {}
 }
 
 service World {

protos/group.proto

@@ -0,0 +1,24 @@
+syntax = "proto3";
+
+import "common.proto";
+
+package dcs.group;
+
+enum GroupCategory {
+  AIRPLANE = 0;
+  HELICOPTER = 1;
+  GROUND = 2;
+  SHIP = 3;
+  TRAIN = 4;
+}
+
+message GetUnitsRequest {
+  string group_name = 1;
+  // Whether the response should include only active units (`true`), only inactive units (`false`),
+  // or all units (`nil`).
+  optional bool active = 2;
+}
+
+message GetUnitsResponse {
+  repeated Unit units = 1;
+}

protos/event_stream.proto → protos/mission.proto

@@ -232,4 +232,33 @@ message Event {
     MarkChangeEvent mark_change = 26;
     MarkRemoveEvent mark_remove = 27;
   }
-}
+}
+
+message StreamUnitsRequest {
+  // The poll rate in seconds at which the gRPC server communicates with the DCS mission to retrieve
+  // the latest unit positions. The lower the `poll_rate` the higher the amount of requests send to
+  // to the DCS mission.
+  // Default: 5
+  optional uint32 poll_rate = 1;
+
+  // The maximum backoff in seconds which the gRPC postpones polling units that haven't moved
+  // recently. This is an optimization to dynamically reduce the poll rate for stationary units.
+  // Set it to the same value as `poll_rate` to disable the backoff.
+  // Default: 30
+  optional uint32 max_backoff = 2;
+}
+
+message UnitUpdate {
+  message UnitGone {
+    uint32 id = 1;
+    string name = 2;
+  }
+
+  oneof update {
+    // The unit is either new or its position changed.
+    Unit unit = 1;
+
+    // The unit does not exist anymore.
+    UnitGone gone = 2;
+  }
+}

protos/unit.proto

@@ -4,7 +4,7 @@ import "common.proto";
 
 package dcs;
 
-message GetRadarRequest {
+message UnitName {
   string name = 1;
 }
 
@@ -19,4 +19,12 @@ message GetRadarResponse {
     Airbase airbase = 7;
     Cargo cargo = 8;
   }
-}
+}
+
+message GetUnitPositionResponse {
+  Position position = 1;
+}
+
+message GetUnitPlayerNameResponse {
+  optional string player_name = 1;
+}

src/lib.rs

@@ -1,9 +1,10 @@
 #![allow(dead_code)]
 #![recursion_limit = "256"]
 
-mod rpc;
+pub mod rpc;
 mod server;
 mod shutdown;
+mod stream;
 
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, RwLock};