client-cpp  0.10.0
LogCollector.hpp
Go to the documentation of this file.
1 /*
2  * Copyright 2014-2016 CyberVision, Inc.
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  * http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16 
17 #ifndef LOGCOLLECTOR_HPP_
18 #define LOGCOLLECTOR_HPP_
19 
20 
21 #include <chrono>
22 #include <memory>
23 #include <future>
24 #include <list>
25 #include <unordered_map>
26 #include <cstdint>
27 
28 #include "kaa/KaaThread.hpp"
29 #include "kaa/log/ILogStorage.hpp"
35 #include "kaa/utils/KaaTimer.hpp"
37 
38 namespace kaa {
39 
40 class IExecutorContext;
41 class LoggingTransport;
43 
44 typedef std::chrono::system_clock clock_t;
45 
46 class TimeoutInfo {
47 
48 public:
49  TimeoutInfo(const std::int32_t& transportAccessPointId, const std::chrono::time_point<clock_t>& timeoutTime)
50  : transportAccessPointId_(transportAccessPointId), timeoutTime_(timeoutTime) {}
51 
52  std::int32_t getTransportAccessPointId() const {
53  return transportAccessPointId_;
54  }
55 
56  std::chrono::time_point<clock_t> getTimeoutTime() const {
57  return timeoutTime_;
58  }
59 
60 private:
61  std::int32_t transportAccessPointId_;
62  std::chrono::time_point<clock_t> timeoutTime_;
63 };
64 
69 public:
71 
72  virtual RecordFuture addLogRecord(const KaaUserLogRecord& record);
73 
74  virtual void setStorage(ILogStoragePtr storage);
75  virtual void setUploadStrategy(ILogUploadStrategyPtr strategy);
76 
78  logDeliverylistener_ = listener;
79  }
80 
81  virtual std::shared_ptr<LogSyncRequest> getLogUploadRequest();
82  virtual void onLogUploadResponse(const LogSyncResponse& response, std::size_t deliveryTime);
83 
84  void setTransport(LoggingTransport* transport) {
85  transport_ = transport;
86  }
87 
88 private:
89  typedef std::shared_ptr<std::promise<RecordInfo>> DeliveryFuture;
90 
91  struct RecordDeliveryInfo {
92  RecordDeliveryInfo(const DeliveryFuture& f, const RecordInfo& info)
93  : deliveryFuture_(f), recordInfo_(info) {}
94 
95  DeliveryFuture deliveryFuture_;
96  RecordInfo recordInfo_;
97  };
98 
99  struct BucketWrapper {
100  BucketInfo bucketInfo_;
101  std::list<RecordDeliveryInfo> recordDeliveryInfoStorage_;
102  };
103 
104 private:
105  virtual void retryLogUpload();
106  virtual void retryLogUpload(std::size_t delay);
107 
108  virtual void switchAccessPoint();
109 
110  void doSync();
111  void processLogUploadDecision(LogUploadStrategyDecision decision);
112 
113  bool isDeliveryTimeout();
114  void addDeliveryTimeout(std::int32_t requestId);
115  bool removeDeliveryTimeout(std::int32_t requestId);
116 
117  void startTimeoutTimer();
118  void startLogUploadCheckTimer();
119 
120  void processTimeout();
121 
122  void rescheduleTimers();
123 
124  bool isUploadAllowed();
125 
126 
127  void updateBucketInfo(const BucketInfo& bucketInfo, const RecordDeliveryInfo& recordInfo);
128  BucketInfo getBucketInfo(std::int32_t);
129  void notifyDeliveryFuturesOnSuccess(std::int32_t bucketId, std::size_t deliveryTime);
130  void removeBucketInfo(std::int32_t);
131 
132 private:
133  ILogStoragePtr storage_;
134  ILogUploadStrategyPtr uploadStrategy_;
135  LoggingTransport* transport_;
136 
137  IKaaChannelManagerPtr channelManager_;
138 
139  std::unordered_map<std::int32_t, TimeoutInfo> timeouts_;
140  std::int32_t timeoutAccessPointId_;
141  KAA_MUTEX_DECLARE(timeoutsGuard_);
142 
143  KaaTimer<void ()> logUploadCheckTimer_;
144  KaaTimer<void ()> scheduledUploadTimer_;
145  KaaTimer<void ()> timeoutTimer_;
146 
147  ILogDeliveryListenerPtr logDeliverylistener_;
148 
149  std::unordered_map<std::int32_t, BucketWrapper> bucketInfoStorage_;
150  KAA_MUTEX_DECLARE(bucketInfoStorageGuard_);
151 
152  IKaaClientContext &context_;
153 };
154 
155 } // namespace kaa
156 
157 #endif /* LOGCOLLECTOR_HPP_ */
virtual void setStorage(ILogStoragePtr storage)
Sets the new log storage.
std::shared_ptr< ILogDeliveryListener > ILogDeliveryListenerPtr
void setTransport(LoggingTransport *transport)
virtual void setUploadStrategy(ILogUploadStrategyPtr strategy)
Sets the new log upload strategy.
std::shared_ptr< ILogStorage > ILogStoragePtr
Definition: ILogStorage.hpp:93
virtual std::shared_ptr< LogSyncRequest > getLogUploadRequest()
IKaaChannelManager * IKaaChannelManagerPtr
TimeoutInfo(const std::int32_t &transportAccessPointId, const std::chrono::time_point< clock_t > &timeoutTime)
The public interface to the Kaa log collecting subsystem.
virtual void onLogUploadResponse(const LogSyncResponse &response, std::size_t deliveryTime)
Describes unique log record delivery info.
Definition: RecordInfo.hpp:30
NOTE: THIS FILE IS AUTO-GENERATED. DO NOT EDIT IT MANUALLY.
LogCollector(IKaaChannelManagerPtr manager, IKaaClientContext &context)
std::shared_ptr< ILogUploadStrategy > ILogUploadStrategyPtr
std::chrono::system_clock clock_t
virtual void setLogDeliveryListener(ILogDeliveryListenerPtr listener)
Set a listener which receives a delivery status of each log bucket.
LogUploadStrategyDecision
Log upload decisions.
std::chrono::time_point< clock_t > getTimeoutTime() const
virtual RecordFuture addLogRecord(const KaaUserLogRecord &record)
Adds a new log record to the log storage.
std::int32_t getTransportAccessPointId() const