WPILibC++  unspecified
 All Classes Files Functions Variables Typedefs Enumerations Enumerator Pages
NetworkConnection.h
1 /*----------------------------------------------------------------------------*/
2 /* Copyright (c) FIRST 2015. All Rights Reserved. */
3 /* Open Source Software - may be modified and shared by FRC teams. The code */
4 /* must be accompanied by the FIRST BSD license file in the root directory of */
5 /* the project. */
6 /*----------------------------------------------------------------------------*/
7 
8 #ifndef NT_NETWORKCONNECTION_H_
9 #define NT_NETWORKCONNECTION_H_
10 
11 #include <atomic>
12 #include <chrono>
13 #include <memory>
14 #include <thread>
15 
16 #include "support/ConcurrentQueue.h"
17 #include "Message.h"
18 #include "ntcore_cpp.h"
19 
20 namespace wpi {
21 class NetworkStream;
22 }
23 
24 namespace nt {
25 
26 class Notifier;
27 
29  public:
30  enum State { kCreated, kInit, kHandshake, kSynchronized, kActive, kDead };
31 
32  typedef std::function<bool(
33  NetworkConnection& conn,
34  std::function<std::shared_ptr<Message>()> get_msg,
35  std::function<void(llvm::ArrayRef<std::shared_ptr<Message>>)> send_msgs)>
36  HandshakeFunc;
37  typedef std::function<void(std::shared_ptr<Message> msg,
38  NetworkConnection* conn)>
39  ProcessIncomingFunc;
40  typedef std::vector<std::shared_ptr<Message>> Outgoing;
41  typedef wpi::ConcurrentQueue<Outgoing> OutgoingQueue;
42 
43  NetworkConnection(std::unique_ptr<wpi::NetworkStream> stream,
44  Notifier& notifier, HandshakeFunc handshake,
45  Message::GetEntryTypeFunc get_entry_type);
47 
48  // Set the input processor function. This must be called before Start().
49  void set_process_incoming(ProcessIncomingFunc func) {
50  m_process_incoming = func;
51  }
52 
53  void Start();
54  void Stop();
55 
56  ConnectionInfo info() const;
57 
58  bool active() const { return m_active; }
59  wpi::NetworkStream& stream() { return *m_stream; }
60 
61  void QueueOutgoing(std::shared_ptr<Message> msg);
62  void PostOutgoing(bool keep_alive);
63  void NotifyIfActive(ConnectionListenerCallback callback) const;
64 
65  unsigned int uid() const { return m_uid; }
66 
67  unsigned int proto_rev() const { return m_proto_rev; }
68  void set_proto_rev(unsigned int proto_rev) { m_proto_rev = proto_rev; }
69 
70  State state() const;
71  void set_state(State state);
72 
73  std::string remote_id() const;
74  void set_remote_id(StringRef remote_id);
75 
76  unsigned long long last_update() const { return m_last_update; }
77 
78  NetworkConnection(const NetworkConnection&) = delete;
79  NetworkConnection& operator=(const NetworkConnection&) = delete;
80 
81  private:
82  void ReadThreadMain();
83  void WriteThreadMain();
84 
85  static std::atomic_uint s_uid;
86 
87  unsigned int m_uid;
88  std::unique_ptr<wpi::NetworkStream> m_stream;
89  Notifier& m_notifier;
90  OutgoingQueue m_outgoing;
91  HandshakeFunc m_handshake;
92  Message::GetEntryTypeFunc m_get_entry_type;
93  ProcessIncomingFunc m_process_incoming;
94  std::thread m_read_thread;
95  std::thread m_write_thread;
96  std::atomic_bool m_active;
97  std::atomic_uint m_proto_rev;
98  mutable std::mutex m_state_mutex;
99  State m_state;
100  mutable std::mutex m_remote_id_mutex;
101  std::string m_remote_id;
102  std::atomic_ullong m_last_update;
103  std::chrono::steady_clock::time_point m_last_post;
104 
105  std::mutex m_pending_mutex;
106  Outgoing m_pending_outgoing;
107  std::vector<std::pair<std::size_t, std::size_t>> m_pending_update;
108 
109  // Condition variables for shutdown
110  std::mutex m_shutdown_mutex;
111  std::condition_variable m_read_shutdown_cv;
112  std::condition_variable m_write_shutdown_cv;
113  bool m_read_shutdown = false;
114  bool m_write_shutdown = false;
115 };
116 
117 } // namespace nt
118 
119 #endif // NT_NETWORKCONNECTION_H_
NetworkTables Connection Information.
Definition: ntcore_cpp.h:43
Definition: NetworkConnection.h:28
Definition: Notifier.h:19