puai_dispatch/snap7_communication/snap7_communication_base.cpp

//
// Created by huli on 2020/9/25.
//

#include "snap7_communication_base.h"
#include "../tool/proto_tool.h"

Snap7_communication_base::Snap7_communication_base()
{
	m_communication_status = E_UNKNOWN;
	m_communication_delay_time_ms = SNAP7_COMMUNICATION_DELAY_TIME_MS;
	mp_communication_thread = NULL;
}

Snap7_communication_base::~Snap7_communication_base()
{
	communication_uninit();
}

//初始化 通信 模块。如下三选一
Error_manager Snap7_communication_base::communication_init()
{
	return  communication_init_from_protobuf(SNAP7_COMMUNICATION_PARAMETER_PATH);
}
//初始化 通信 模块。从文件读取
Error_manager Snap7_communication_base::communication_init_from_protobuf(std::string prototxt_path)
{
	Snap7_communication_proto::Snap7_communication_parameter_all t_snap7_communication_parameter_all;
	if(!  proto_tool::read_proto_param(prototxt_path,t_snap7_communication_parameter_all) )
	{
		return Error_manager(SNAP7_READ_PROTOBUF_ERROR,MINOR_ERROR,
							 "Snap7_communication_base::communication_init_from_protobuf read_proto_param  failed");
	}
	return communication_init_from_protobuf(t_snap7_communication_parameter_all);
}
//初始化 通信 模块。从protobuf读取
Error_manager Snap7_communication_base::communication_init_from_protobuf(Snap7_communication_proto::Snap7_communication_parameter_all& snap7_communication_parameter_all)
{
	LOG(INFO) << " ---Communication_socket_base::communication_init() --- "<< this;
	Error_manager t_error;
//	snap7_communication_parameter_all.DebugString();

	if ( snap7_communication_parameter_all.snap7_communication_parameters().has_ip_string() )
	{
		m_ip_string = snap7_communication_parameter_all.snap7_communication_parameters().ip_string();
		t_error = communication_connect(m_ip_string);
		if ( t_error != Error_code::SUCCESS )
		{
			//连接失败, 不要直接返回, 而是改为断连, 后面继续启动线程, (线程内部有重连功能)
			m_communication_status = E_DISCONNECT;
		}
		else
		{
			m_communication_status = E_READY;
		}
	}

	//启动通信, run thread
	communication_run();
	return Error_code::SUCCESS;
}


//反初始化 通信 模块。
Error_manager Snap7_communication_base::communication_uninit()
{
	//关闭线程并回收资源
	if (mp_communication_thread)
	{
		m_communication_condition.kill_all();
	}
	if (mp_communication_thread)
	{
		mp_communication_thread->join();
		delete mp_communication_thread;
		mp_communication_thread = NULL;
	}

	//清空map
	{
		std::unique_lock<std::mutex> t_lock(m_receive_buf_lock);
		m_receive_buf_map.clear();
	}
	{
		std::unique_lock<std::mutex> t_lock(m_send_buf_lock);
		m_send_buf_map.clear();
	}

	communication_disconnect();
	m_communication_status = E_UNKNOWN;

	return Error_code::SUCCESS;
}

Snap7_communication_base::Communication_statu Snap7_communication_base::get_status()
{
	return m_communication_status;
}

//通信连接
Error_manager Snap7_communication_base::communication_connect(std::string ip_string)
{
	std::unique_lock<std::mutex> t_lock(m_communication_lock);
	int result=m_snap7_client.ConnectTo(ip_string.c_str(),0,1);
	std::this_thread::sleep_for(std::chrono::milliseconds(m_communication_delay_time_ms));
	if (result==0)
	{
		return Error_code::SUCCESS;
	}
	else
	{
		return Error_manager(Error_code::SNAP7_CONNECT_ERROR, Error_level::MINOR_ERROR,
							 " Snap7_communication_base::communication_connect error ");
	}
	return Error_code::SUCCESS;
}
//启动通信, run thread
Error_manager Snap7_communication_base::communication_run()
{
	//启动4个线程。
	//接受线程默认循环, 内部的nn_recv进行等待, 超时1ms
	m_communication_condition.reset(false, false, false);
	mp_communication_thread = new std::thread(&Snap7_communication_base::communication_thread, this);

	return Error_code::SUCCESS;
}
//通信断连
Error_manager Snap7_communication_base::communication_disconnect()
{
	std::unique_lock<std::mutex> t_lock(m_communication_lock);
	int result=m_snap7_client.Disconnect();
	std::this_thread::sleep_for(std::chrono::milliseconds(m_communication_delay_time_ms));
	if (result==0)
	{
		return Error_code::SUCCESS;
	}
	else
	{
		return Error_manager(Error_code::SNAP7_DISCONNECT_ERROR, Error_level::MINOR_ERROR,
							 " Snap7_communication_base::communication_disconnect error ");
	}
	return Error_code::SUCCESS;
}
//mp_communication_thread线程的执行函数, 负责进行s7的通信
void Snap7_communication_base::communication_thread()
{
	LOG(INFO) << " ---Snap7_communication_base::communication_thread()--- "<< this;
Error_manager t_error;
	while (m_communication_condition.is_alive())
	{
		m_communication_condition.wait_for_ex(std::chrono::milliseconds(m_communication_delay_time_ms));

		//s7的通信时间较长, 所以将发送和接受分开
		//发送多个时, 必须加锁后一起发送, 不允许分段写入, 防止数据错误
		if ( m_communication_condition.is_alive() )
		{
			std::this_thread::yield();
			switch ( m_communication_status )
			{
			    case E_READY:
				case E_RECEIVE:
			    {
					{
						std::unique_lock<std::mutex> t_lock(m_receive_buf_lock);
						for (auto iter = m_receive_buf_map.begin(); iter != m_receive_buf_map.end(); ++iter)
						{
							//接受数据, 读取DB块,
							t_error = read_data_buf(iter->second);
							if (t_error == Error_code::SNAP7_READ_ERROR)
							{
								m_communication_status = E_DISCONNECT;
								std::cout << " huli test :::: " << " t_error = " << t_error << std::endl;
								break;
							}
						}
					}
					//注:数据更新放在锁的外面, 防止重复加锁....
					updata_receive_buf();
					m_communication_status = E_SEND;
			        break;
			    }
			    case E_SEND:
			    {
					//注:数据更新放在锁的外面, 防止重复加锁....
					updata_send_buf();
					{
						std::unique_lock<std::mutex> t_lock(m_send_buf_lock);
						for (auto iter = m_send_buf_map.begin(); iter != m_send_buf_map.end(); ++iter)
						{
							//发送数据, 写入DB块,
							t_error = write_data_buf(iter->second);
							if (t_error == Error_code::SNAP7_WRITE_ERROR)
							{
								m_communication_status = E_DISCONNECT;
								std::cout << " huli test :::: " << " t_error = " << t_error << std::endl;
								break;
							}
						}
					}
					m_communication_status = E_RECEIVE;
			        break;
			    }
				case E_DISCONNECT:
				{
					//重连
					LOG(WARNING) << "find plc connection error, trying to reconnect.";
					communication_disconnect();
					std::this_thread::sleep_for(std::chrono::milliseconds(m_communication_delay_time_ms));
					t_error = communication_connect(m_ip_string);
					if ( t_error != Error_code::SUCCESS )
					{
						//连接失败, 不要直接返回, 而是改为断连, 后面继续启动线程, (线程内部有重连功能)
						m_communication_status = E_DISCONNECT;
					}
					else
					{
						m_communication_status = E_READY;
					}
					std::this_thread::sleep_for(std::chrono::milliseconds(m_communication_delay_time_ms));

					break;
				}
			    default:
			    {

			        break;
			    }
			}
		}
	}

	LOG(INFO) << " Communication_socket_base::send_data_thread end "<< this;
	return;
}

//接受数据, 读取DB块,
Error_manager Snap7_communication_base::read_data_buf(Snap7_buf& snap7_buf)
{
	Error_manager t_error;
	if ( snap7_buf.m_communication_mode != Snap7_buf::NO_COMMUNICATION)
	{
		if ( snap7_buf.m_communication_mode == Snap7_buf::ONCE_COMMUNICATION )
		{
			snap7_buf.m_communication_mode = Snap7_buf::NO_COMMUNICATION;
		}

		std::unique_lock<std::mutex> lck(m_communication_lock);
		int result = m_snap7_client.AsDBRead(snap7_buf.m_id, snap7_buf.m_start_index, snap7_buf.m_size, snap7_buf.mp_buf_reverse);
		if ( result == 0 )
		{
			m_snap7_client.WaitAsCompletion(100);
			//倒序数据 转为 正序数据
			snap7_buf.reverse_to_obverse();
		}
		else
		{
			return Error_manager(Error_code::SNAP7_READ_ERROR, Error_level::MINOR_ERROR,
								 " Snap7_communication_base::read_data_buf error ");
		}
	}

	return Error_code::SUCCESS;
}
//发送数据, 写入DB块,
Error_manager Snap7_communication_base::write_data_buf(Snap7_buf& snap7_buf)
{
	Error_manager t_error;
	if ( snap7_buf.m_communication_mode != Snap7_buf::NO_COMMUNICATION)
	{
		if ( snap7_buf.m_communication_mode != Snap7_buf::ONCE_COMMUNICATION )
		{
			snap7_buf.m_communication_mode = Snap7_buf::NO_COMMUNICATION;
		}

		//正序数据 转为 倒序数据
		snap7_buf.obverse_to_reverse();

		std::unique_lock<std::mutex> lck(m_communication_lock);
		int result = m_snap7_client.AsDBWrite(snap7_buf.m_id, snap7_buf.m_start_index, snap7_buf.m_size,
											  snap7_buf.mp_buf_reverse);
		if ( result == 0 )
		{
			m_snap7_client.WaitAsCompletion(100);
		}
		else
		{
			return Error_manager(Error_code::SNAP7_WRITE_ERROR, Error_level::MINOR_ERROR,
								 " Snap7_communication_base::write_data_buf error ");
		}
	}
	return Error_code::SUCCESS;
}

//数据颠倒
Error_manager Snap7_communication_base::reverse_byte(void* p_buf_in, void* p_buf_out, int size)
{
	if ( p_buf_in == NULL || p_buf_out == NULL )
	{
	    return Error_manager(Error_code::POINTER_IS_NULL, Error_level::MINOR_ERROR,
	    					"  POINTER IS NULL ");
	}
	char* tp_in=(char*)p_buf_in;
	char* tp_out=(char*)p_buf_out;
//	for(int i=0;i<size;++i)
//	{
//		tp_out[i]=tp_in[size-i-1];
//	}
	for(int i=0;i<size;++i)
	{
		tp_out[i]=tp_in[i];
	}
	return Error_code::SUCCESS;
}

//更新数据
Error_manager Snap7_communication_base::updata_receive_buf()
{
	return Error_code::SUCCESS;
}
Error_manager Snap7_communication_base::updata_send_buf()
{
	return Error_code::SUCCESS;
}