Drivers: add eph and obs writer in rtcm parser

modules/drivers/gnss/parser/rtcm_parser.cc

@@ -32,6 +32,11 @@ namespace gnss {
 using ::apollo::drivers::gnss::EpochObservation;
 using ::apollo::drivers::gnss::GnssEphemeris;
 
+RtcmParser::RtcmParser(const config::Config &config,
+           const std::shared_ptr<Node> &node)
+    : config_(config), node_(node) {
+}
+
 bool RtcmParser::Init() {
   rtcm_parser_.reset(new Rtcm3Parser(true));
 
@@ -40,6 +45,11 @@ bool RtcmParser::Init() {
     return false;
   }
 
+  gnssephemeris_writer_ =
+      node_->CreateWriter<GnssEphemeris>(config_.gnssephemeris_channel_name());
+  epochobservation_writer_ = node_->CreateWriter<EpochObservation>(
+      config_.epochobservation_channel_name());
+
   inited_flag_ = true;
   return true;
 }
@@ -75,14 +85,15 @@ void RtcmParser::DispatchMessage(Parser::MessageType type, MessagePtr message) {
   }
 }
 
-void RtcmParser::PublishEphemeris(const MessagePtr message) {
-  GnssEphemeris eph = *As<GnssEphemeris>(message);
-  // AdapterManager::PublishGnssRtkEph(eph);
+void RtcmParser::PublishEphemeris(const MessagePtr& message) {
+  auto eph = std::make_shared<GnssEphemeris>(*As<GnssEphemeris>(message));
+  gnssephemeris_writer_->Write(eph);
 }
 
-void RtcmParser::PublishObservation(const MessagePtr message) {
-  EpochObservation observation = *As<EpochObservation>(message);
-  // AdapterManager::PublishGnssRtkObs(observation);
+void RtcmParser::PublishObservation(const MessagePtr& message) {
+  auto observation =
+      std::make_shared<EpochObservation>(*As<EpochObservation>(message));
+  epochobservation_writer_->Write(observation);
 }
 
 }  // namespace gnss

modules/drivers/gnss/parser/rtcm_parser.h

@@ -24,24 +24,38 @@
 
 #include "modules/drivers/gnss/parser/parser.h"
 
+#include "modules/drivers/gnss/proto/gnss_raw_observation.pb.h"
+
 namespace apollo {
 namespace drivers {
 namespace gnss {
 
+using ::apollo::drivers::gnss::EpochObservation;
+using ::apollo::drivers::gnss::GnssEphemeris;
+
+using apollo::cybertron::Node;
+using apollo::cybertron::Reader;
+using apollo::cybertron::Writer;
+
 class RtcmParser {
  public:
-  RtcmParser() {}
+  RtcmParser(const config::Config &config,
+             const std::shared_ptr<Node> &node);
   ~RtcmParser() {}
   bool Init();
   void ParseRtcmData(const std::string& msg);
 
  private:
   void DispatchMessage(Parser::MessageType type, MessagePtr message);
-  void PublishEphemeris(const MessagePtr message);
-  void PublishObservation(const MessagePtr message);
+  void PublishEphemeris(const MessagePtr& message);
+  void PublishObservation(const MessagePtr& message);
 
+  config::Config config_;
+  std::shared_ptr<Node> node_ = nullptr;
+  std::shared_ptr<Writer<GnssEphemeris>> gnssephemeris_writer_ = nullptr;
+  std::shared_ptr<Writer<EpochObservation>> epochobservation_writer_ = nullptr;
   bool inited_flag_ = false;
-  std::unique_ptr<Parser> rtcm_parser_;
+  std::unique_ptr<Parser> rtcm_parser_ = nullptr;
 };
 
 }  // namespace gnss

modules/drivers/gnss/stream/raw_stream.cc

@@ -134,7 +134,7 @@ RawStream::RawStream(const config::Config &config,
                      const std::shared_ptr<Node> &node)
     : config_(config), node_(node) {
   data_parser_ptr_.reset(new DataParser(config_, node_));
-  rtcm_parser_ptr_.reset(new RtcmParser());
+  rtcm_parser_ptr_.reset(new RtcmParser(config_, node_));
 }
 
 RawStream::~RawStream() {