1 /*
2 * Copyright (c) 2021 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15 #include "stream_plugin.h"
16 #include "securec.h"
17
18 #include <sys/syscall.h>
19 #include <sys/types.h>
20 #include <unistd.h>
21
StreamPlugin()22 StreamPlugin::StreamPlugin() {}
23
~StreamPlugin()24 StreamPlugin::~StreamPlugin() {}
25
Start(const uint8_t * configData,uint32_t configSize)26 int StreamPlugin::Start(const uint8_t* configData, uint32_t configSize)
27 {
28 // 反序列化
29 if (protoConfig_.ParseFromArray(configData, configSize) <= 0) {
30 HILOG_ERROR(LOG_CORE, "%s:parseFromArray failed!", __func__);
31 return -1;
32 }
33 // 启动线程写数据
34 std::unique_lock<std::mutex> locker(mutex_);
35 running_ = true;
36 writeThread_ = std::thread(&StreamPlugin::Loop, this);
37
38 return 0;
39 }
40
Stop()41 int StreamPlugin::Stop()
42 {
43 std::unique_lock<std::mutex> locker(mutex_);
44 running_ = false;
45 locker.unlock();
46 if (writeThread_.joinable()) {
47 writeThread_.join();
48 }
49 HILOG_INFO(LOG_CORE, "%s:stop success!", __func__);
50 return 0;
51 }
52
SetWriter(WriterStruct * writer)53 int StreamPlugin::SetWriter(WriterStruct* writer)
54 {
55 resultWriter_ = writer;
56 return 0;
57 }
58
GetTimeMS()59 uint64_t StreamPlugin::GetTimeMS()
60 {
61 const int MS_PER_S = 1000;
62 const int NS_PER_MS = 1000000;
63 struct timespec ts;
64 clock_gettime(CLOCK_BOOTTIME, &ts);
65 return ts.tv_sec * MS_PER_S + ts.tv_nsec / NS_PER_MS;
66 }
67
Loop(void)68 void StreamPlugin::Loop(void)
69 {
70 HILOG_INFO(LOG_CORE, "%s:transporter thread %d start !!!!!", __func__, gettid());
71 uint32_t index = 0;
72 while (running_) {
73 StreamData dataProto;
74 dataProto.set_intdata(index);
75 dataProto.set_stringdata(std::to_string(index));
76 buffer_.resize(dataProto.ByteSizeLong());
77 dataProto.SerializeToArray(buffer_.data(), buffer_.size());
78
79 if (index < 50) {
80 if (resultWriter_->write != nullptr) {
81 resultWriter_->write(resultWriter_, buffer_.data(), buffer_.size());
82 resultWriter_->flush(resultWriter_);
83 }
84 }
85 index++;
86 }
87 resultWriter_->flush(resultWriter_);
88 HILOG_INFO(LOG_CORE, "%s:transporter thread %d exit !!!!!", __func__, gettid());
89 }