Commit 0592132f authored by liuyu's avatar liuyu

udp receive ok

parent 8cf85807
......@@ -52,7 +52,7 @@ void ngap_app::handle_receive(
"Handling SCTP payload from SCTP Server on assoc_id (%d), stream_id "
"(%d), instreams (%d), outstreams (%d)",
assoc_id, stream, instreams, outstreams);
bool priority = false;
Ngap_NGAP_PDU_t* ngap_msg_pdu =
(Ngap_NGAP_PDU_t*) calloc(1, sizeof(Ngap_NGAP_PDU_t));
asn_dec_rval_t rc = asn_decode(
......@@ -75,9 +75,42 @@ void ngap_app::handle_receive(
}
// Handle the message
if(ngap_msg_pdu->choice.initiatingMessage->procedureCode != 25 && priority == true
&& ngap_msg_pdu->choice.initiatingMessage->procedureCode != 21)
{
Logger::ngap().info("starting udp send");
Logger::ngap().debug("assoc_id = %d",assoc_id);
// for(int i = 0; i < n; i++)
// {
// std::stringstream data;
// data << std::hex << (int)buffer[i];
// std::cout << data.str();
// }
// std::cout<<std::endl;
// buffer[n] = assoc_id;
// buffer[n+1] = stream;
// for(int i = 0; i < n+2; i++)
// {
// std::stringstream data;
// data << std::hex << (int)buffer[i];
// std::cout << data.str();
// }
// std::cout<<std::endl;
// // std::string ngapmsg;
// // octet_stream_2_hex_stream2((uint8_t*) bdata(payload), blength(payload), ngapmsg);
// //curl_http_another_amf(assoc_id,stream,ngapmsg);
// UdpSend(buffer,n+2,"10.103.238.72",9000);
}
else{
//Handle the message
(*messages_callback[ngap_msg_pdu->choice.initiatingMessage->procedureCode]
[ngap_msg_pdu->present - 1])(
assoc_id, stream, ngap_msg_pdu);
}
}
//------------------------------------------------------------------------------
......
#include "ngap_app.hpp"
#include "logger.hpp"
#include <nlohmann/json.hpp>
#include <curl/curl.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/time.h>
#include <unistd.h>
// #include "amf_sbi.hpp"
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <string>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <signal.h>
#include <iostream>
#include "udp.hpp"
#include <thread>
#include "bstrlib.h"
#include "sstream"
extern "C" {
#include "Ngap_InitiatingMessage.h"
#include "Ngap_NGAP-PDU.h"
}
using namespace std;
udp_server::udp_server(const char* udp_address,uint16_t udp_port_num){
cout<<"start create socket"<<endl;
create_socket(udp_address,udp_port_num);
cout<<"start receive"<<endl;
start_receive(udp_);
}
udp_server::~udp_server(){}
int udp_server::create_socket(const char* udp_address,uint16_t udp_port_num){
//1. 创建socket 参数一: 协议类型(版本) 参数二: 通信媒介 参数三: 保护方式
cout<<"start"<<endl;
serverSocket = socket(AF_INET,SOCK_DGRAM,IPPROTO_UDP);
if(-1 == serverSocket) printf("创建socket失败:%m\n"),exit(-1);
printf("创建socket成功!\n");
//2. 创建服务器协议地址簇
struct sockaddr_in sAddr = {0};
//uint16_t port = 8000;
//const char* ip_address= "10.103.239.24";
sAddr.sin_family = AF_INET; //协议类型 和socket函数第一个参数一致
sAddr.sin_addr.s_addr = inet_addr(udp_address); //将字符串转整数
sAddr.sin_port = htons(udp_port_num); //整数转整数 小端转大端
//3.绑定
int r = bind(serverSocket,(struct sockaddr*)&sAddr,sizeof sAddr);
if(-1 == r) printf("绑定失败:%m\n"),close(serverSocket),exit(-2);
printf("绑定成功!\n");
return serverSocket;
// struct sockaddr_in addr = {};
// int sd = 0;
// /*
// * Create UDP socket
// */
// if ((sd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) {
// /*
// * Socket creation has failed...
// */
// printf("Socket creation failed (%s)\n", strerror(errno));
// return errno;
// }
// addr.sin_family = AF_INET;
// addr.sin_port = htons(udp_port_num);
// addr.sin_addr.s_addr = inet_addr(udp_address);
// std::string ipv4 = udp_address;
// printf(
// "Creating new listen socket on address %s and port %" PRIu16 " ",
// ipv4.c_str(), udp_port_num);
// if (bind(sd, (struct sockaddr*) &addr, sizeof(struct sockaddr_in)) < 0) {
// /*
// * Bind failed
// */
// printf(
// "Socket bind failed (%s) for address %s and port %" PRIu16 " \n",
// strerror(errno), ipv4.c_str(), udp_port_num);
// close(sd);
// return errno;
// }
}
void udp_server::start_receive(udp_server* udp){
udp_ = udp;
//pthread_create(&udp_thread, NULL, udp_receiver_thread, (void*) this);
//pthread_exit(NULL);
thread_ = std::thread(&udp_server::udp_read_from_socket,this);
thread_.detach();
}
// void* udp_server::udp_receiver_thread(void* arg) {
// udp_server* ptr = (udp_server*) arg;
// //Logger::udp().info("Create pthread to receive udp message");
// ptr->udp_read_from_socket();
// }
int udp_server::getSocket(){
return serverSocket;
}
int udp_server::getPpid(){
return ppid;
}
int udp_server::udp_read_from_socket() {
//4.通信
uint8_t udp_recv[UDP_RECV_BUFFER_SIZE];
//bstring udp_recv;
//struct sockaddr_in cAddr = {0};
//socklen_t len = sizeof(cAddr);
endpoint r_endpoint = {};
//socklen_t addr_len;
int udp_len;
size_t bytes_received = 0;
cout<<"start receive"<<endl;
assoc_id = 0;
stream = 0;
struct timeval time;
while(1){
r_endpoint.addr_storage_len=sizeof(struct sockaddr_storage);
//如果还需要向客户端发送东西用recvfrom
//udp精髓 向一个协议地址簇发东西
//收消息
//r = recvfrom(serverSocket,udp_recv,udp_len,0,(struct sockaddr*)&cAddr,&len);
bytes_received = recvfrom(serverSocket, (void *)udp_recv, UDP_RECV_BUFFER_SIZE , 0, (struct sockaddr*)&r_endpoint.addr_storage, &r_endpoint.addr_storage_len);
//bytes_received = recvfrom(serverSocket, udp_recv, UDP_RECV_BUFFER_SIZE , 0, (struct sockaddr*)&r_endpoint.addr_storage, &r_endpoint.addr_storage_len);
if(bytes_received > 0){
printf("receive udp success\n");
gettimeofday(&time, NULL);
std::cout<<"receive from plugin, ms: "<<(time.tv_sec*1000 + time.tv_usec/1000)<<std::endl;
// cout<<"bytes"<<bytes_received<<endl;
// cout<<"content:"<<udp_recv<<endl;
// string udp_recv_str(&udp_recv[0],&udp_recv[strlen(udp_recv)]);
// cout<<"content:"<<udp_recv_str<<endl;
handle_receive_udp(bytes_received,udp_recv);
// assoc_id++;
// stream++;
//ngap_app("0",0).handle_receive(udp_recv_str,assoc_id,stream,0,0);
}
}
return 0;
}
vector<string> SplitString(string s, string c){
string::size_type pos1, pos2;
vector<string> v;
pos2 = s.find(c);
pos1 = 0;
while(string::npos != pos2)
{
v.push_back(s.substr(pos1, pos2-pos1));
pos1 = pos2 + c.size();
pos2 = s.find(c, pos1);
}
if(pos1 != s.length())
v.push_back(s.substr(pos1));
return v;
}
void udp_server::handle_receive_udp(int r,uint8_t udp_recv[]){
stringstream ngap_msg_stream;
sctp_stream_id_t stream;
sctp_assoc_id_t assoc_id;
vector<string> MsgVec;
uint8_t buffer[2048];
//拆分字符串分出连接号
// ngap_msg_stream<<udp_recv;
// ngap_msg_stream>>ngap_msg;
//MsgVec = SplitString(ngap_msg,"|");
//数组赋值
for(int i=0;i<r-2;i++){
buffer[i] = udp_recv[i];
}
int _assoc_id = udp_recv[r-2];
int _stream = udp_recv[r-1];
//打印buffer检查数据
// std::cout<<"buffer: ";
// for(int i = 0; i < r-2; i++)
// {
// std::stringstream data;
// data << std::hex << (int)buffer[i];
// std::cout << data.str();
// }
// std::cout<<std::endl;
assoc_id = _assoc_id;
bstring payload=blk2bstr(buffer,r-2);
// //bstring udp_recv_bstr=bfromcstr(content.c_str());
// cout<<"data:"<<payload->data<<endl;
// cout<<"slen"<<payload->slen<<endl;
// cout<<"mlen"<<payload->mlen<<endl;
cout<<"start handle"<<endl;
// ngap_app("0",0).handle_receive(payload,assoc_id,stream,0,0);
// Ngap_NGAP_PDU_t* ngap_msg_pdu =
// (Ngap_NGAP_PDU_t*) calloc(1, sizeof(Ngap_NGAP_PDU_t));
// Logger::udp().info("start decode");
// asn_dec_rval_t rc = asn_decode(
// NULL, ATS_ALIGNED_CANONICAL_PER, &asn_DEF_Ngap_NGAP_PDU,
// (void**) &ngap_msg_pdu, bdata(udp_recv_bstr), blength(udp_recv_bstr));
// Logger::udp().debug(
// "Decoded UDP message, procedure code %d, present %d",
// ngap_msg_pdu->choice.initiatingMessage->procedureCode,
// ngap_msg_pdu->present);
// asn_fprint(stderr, &asn_DEF_Ngap_NGAP_PDU, ngap_msg_pdu);
// (*messages_callback[ngap_msg_pdu->choice.initiatingMessage->procedureCode]
// [ngap_msg_pdu->present - 1])(
// assoc_id, stream, ngap_msg_pdu);
}
#ifndef __UDP_H
#define __UDP_H
#include "gNB_context.hpp"
#include "sctp_server.hpp"
#include <iostream>
#include <string>
#include <vector>
#include <thread>
#define UDP_RECV_BUFFER_SIZE 8192
using namespace std;
class udp_server{
private:
void* udp_recv;
struct sockaddr_in cAddr = {0};
socklen_t len = sizeof(cAddr);
int udp_len;
uint16_t m_port;
const char* m_address;
int serverSocket;
sctp_assoc_id_t assoc_id;
sctp_stream_id_t stream;
udp_server* udp_;
pthread_t udp_thread;
uint32_t ppid;
std::thread thread_;
public:
udp_server(const char* address,uint16_t port_num);
virtual ~udp_server();
int create_socket(const char* address,uint16_t port_num);
void start_receive(udp_server* udp);
void handle_receive();
static void* udp_receiver_thread(void* arg);
int udp_read_from_socket();
void handle_receive_udp(int r,uint8_t udp_recv[]);
int getSocket();
int getPpid();
//void SplitString(string s, vector<string>& v, string c);
};
#endif
\ No newline at end of file
......@@ -43,6 +43,8 @@
#include "pistache/http.h"
#include "pistache/router.h"
#include "udp.hpp"
using namespace config;
using namespace amf_application;
......@@ -115,6 +117,8 @@ int main(int argc, char** argv) {
Logger::amf_app().debug("Initiating AMF server endpoints");
// AMF HTTP1 server
udp_server udp_("10.103.239.31",9000);
Pistache::Address addr(
std::string(inet_ntoa(*((struct in_addr*) &amf_cfg.sbi.addr4))),
Pistache::Port(amf_cfg.sbi.port));
......@@ -122,6 +126,7 @@ int main(int argc, char** argv) {
amf_api_server_1->init(2);
// std::thread amf_http1_manager(&AMFApiServer::start, amf_api_server_1);
amf_api_server_1->start();
// AMF HTTP2 server
amf_api_server_2 = new amf_http2_server(
conv::toString(amf_cfg.sbi.addr4), amf_cfg.sbi_http2_port, amf_app_inst);
......@@ -131,7 +136,7 @@ int main(int argc, char** argv) {
// amf_http1_manager.join();
// amf_http2_manager.join();
Logger::amf_app().debug("Initiation Done!");
pause();
return 0;
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment