Skip to content
Projects
Groups
Snippets
Help
This project
Loading...
Sign in / Register
Toggle navigation
U
uniset2
Project
Project
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
1
Issues
1
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
UniSet project repositories
uniset2
Commits
02375558
Commit
02375558
authored
Sep 12, 2010
by
Pavel Vainerman
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
UDPExchange: add UDPSender and UDPReceiver
parent
6f4c4f57
Show whitespace changes
Inline
Side-by-side
Showing
11 changed files
with
1519 additions
and
4 deletions
+1519
-4
Makefile.am
extensions/UDPExchange/Makefile.am
+18
-2
UDPReceiver.cc
extensions/UDPExchange/UDPReceiver.cc
+471
-0
UDPReceiver.h
extensions/UDPExchange/UDPReceiver.h
+102
-0
UDPSender.cc
extensions/UDPExchange/UDPSender.cc
+579
-0
UDPSender.h
extensions/UDPExchange/UDPSender.h
+112
-0
start_fg.sh
extensions/UDPExchange/start_fg.sh
+1
-2
start_fg_receiver.sh
extensions/UDPExchange/start_fg_receiver.sh
+8
-0
start_fg_sender.sh
extensions/UDPExchange/start_fg_sender.sh
+8
-0
udpexchange.cc
extensions/UDPExchange/udpexchange.cc
+40
-0
udpreceiver.cc
extensions/UDPExchange/udpreceiver.cc
+92
-0
udpsender.cc
extensions/UDPExchange/udpsender.cc
+88
-0
No files found.
extensions/UDPExchange/Makefile.am
View file @
02375558
bin_PROGRAMS
=
@PACKAGE@-udpexchange
bin_PROGRAMS
=
@PACKAGE@-udpexchange
@PACKAGE@-udpsender @PACKAGE@-udpreceiver
lib_LTLIBRARIES
=
libUniSetUDP.la
lib_LTLIBRARIES
=
libUniSetUDP.la
libUniSetUDP_la_LIBADD
=
$(top_builddir)
/lib/libUniSet.la
\
libUniSetUDP_la_LIBADD
=
$(top_builddir)
/lib/libUniSet.la
\
...
@@ -6,7 +6,9 @@ libUniSetUDP_la_LIBADD = $(top_builddir)/lib/libUniSet.la \
...
@@ -6,7 +6,9 @@ libUniSetUDP_la_LIBADD = $(top_builddir)/lib/libUniSet.la \
$(top_builddir)
/extensions/lib/libUniSetExtensions.la
\
$(top_builddir)
/extensions/lib/libUniSetExtensions.la
\
$(SIGC_LIBS)
$(COMCPP_LIBS)
$(SIGC_LIBS)
$(COMCPP_LIBS)
libUniSetUDP_la_CXXFLAGS
=
-I
$(top_builddir)
/extensions/include
-I
$(top_builddir)
/extensions/SharedMemory
$(SIGC_CFLAGS)
$(COMCPP_CFLAGS)
libUniSetUDP_la_CXXFLAGS
=
-I
$(top_builddir)
/extensions/include
-I
$(top_builddir)
/extensions/SharedMemory
$(SIGC_CFLAGS)
$(COMCPP_CFLAGS)
libUniSetUDP_la_SOURCES
=
UDPPacket.cc UDPExchange.cc
libUniSetUDP_la_SOURCES
=
UDPPacket.cc UDPExchange.cc UDPSender.cc UDPReceiver.cc
#UDPSender.cc
@PACKAGE@
_udpexchange_SOURCES
=
udpexchange.cc
@PACKAGE@
_udpexchange_SOURCES
=
udpexchange.cc
@PACKAGE@
_udpexchange_LDADD
=
libUniSetUDP.la
$(top_builddir)
/lib/libUniSet.la
\
@PACKAGE@
_udpexchange_LDADD
=
libUniSetUDP.la
$(top_builddir)
/lib/libUniSet.la
\
...
@@ -15,6 +17,20 @@ libUniSetUDP_la_SOURCES = UDPPacket.cc UDPExchange.cc
...
@@ -15,6 +17,20 @@ libUniSetUDP_la_SOURCES = UDPPacket.cc UDPExchange.cc
$(SIGC_LIBS)
$(COMCPP_LIBS)
$(SIGC_LIBS)
$(COMCPP_LIBS)
@PACKAGE@
_udpexchange_CXXFLAGS
=
-I
$(top_builddir)
/extensions/include
-I
$(top_builddir)
/extensions/SharedMemory
$(SIGC_CFLAGS)
$(COMCPP_CFLAGS)
@PACKAGE@
_udpexchange_CXXFLAGS
=
-I
$(top_builddir)
/extensions/include
-I
$(top_builddir)
/extensions/SharedMemory
$(SIGC_CFLAGS)
$(COMCPP_CFLAGS)
@PACKAGE@
_udpsender_SOURCES
=
udpsender.cc
@PACKAGE@
_udpsender_LDADD
=
libUniSetUDP.la
$(top_builddir)
/lib/libUniSet.la
\
$(top_builddir)
/extensions/SharedMemory/libUniSetSharedMemory.la
\
$(top_builddir)
/extensions/lib/libUniSetExtensions.la
\
$(SIGC_LIBS)
$(COMCPP_LIBS)
@PACKAGE@
_udpsender_CXXFLAGS
=
-I
$(top_builddir)
/extensions/include
-I
$(top_builddir)
/extensions/SharedMemory
$(SIGC_CFLAGS)
$(COMCPP_CFLAGS)
@PACKAGE@
_udpreceiver_SOURCES
=
udpreceiver.cc
@PACKAGE@
_udpreceiver_LDADD
=
libUniSetUDP.la
$(top_builddir)
/lib/libUniSet.la
\
$(top_builddir)
/extensions/SharedMemory/libUniSetSharedMemory.la
\
$(top_builddir)
/extensions/lib/libUniSetExtensions.la
\
$(SIGC_LIBS)
$(COMCPP_LIBS)
@PACKAGE@
_udpreceiver_CXXFLAGS
=
-I
$(top_builddir)
/extensions/include
-I
$(top_builddir)
/extensions/SharedMemory
$(SIGC_CFLAGS)
$(COMCPP_CFLAGS)
# install
# install
devel_include_HEADERS
=
*
.h
devel_include_HEADERS
=
*
.h
...
...
extensions/UDPExchange/UDPReceiver.cc
0 → 100644
View file @
02375558
// $Id: UDPReceiver.cc,v 1.1 2009/02/10 20:38:27 vpashka Exp $
// -----------------------------------------------------------------------------
#include <sstream>
#include "Exceptions.h"
#include "Extensions.h"
#include "UDPReceiver.h"
// -----------------------------------------------------------------------------
using
namespace
std
;
using
namespace
UniSetTypes
;
using
namespace
UniSetExtensions
;
// -----------------------------------------------------------------------------
UDPReceiver
::
UDPReceiver
(
UniSetTypes
::
ObjectId
objId
,
UniSetTypes
::
ObjectId
shmId
,
SharedMemory
*
ic
)
:
UniSetObject_LT
(
objId
),
shm
(
0
),
initPause
(
0
),
udp
(
0
),
activated
(
false
)
{
if
(
objId
==
DefaultObjectId
)
throw
UniSetTypes
::
SystemError
(
"(UDPReceiver): objId=-1?!! Use --udp-name"
);
// xmlNode* cnode = conf->getNode(myname);
cnode
=
conf
->
getNode
(
myname
);
if
(
cnode
==
NULL
)
throw
UniSetTypes
::
SystemError
(
"(UDPReceiver): Not find conf-node for "
+
myname
);
shm
=
new
SMInterface
(
shmId
,
&
ui
,
objId
,
ic
);
UniXML_iterator
it
(
cnode
);
// определяем фильтр
s_field
=
conf
->
getArgParam
(
"--udp-filter-field"
);
s_fvalue
=
conf
->
getArgParam
(
"--udp-filter-value"
);
dlog
[
Debug
::
INFO
]
<<
myname
<<
"(init): read fileter-field='"
<<
s_field
<<
"' filter-value='"
<<
s_fvalue
<<
"'"
<<
endl
;
// ---------- init RS ----------
// UniXML_iterator it(cnode);
string
s_host
=
conf
->
getArgParam
(
"--udp-host"
,
it
.
getProp
(
"host"
));
if
(
s_host
.
empty
()
)
throw
UniSetTypes
::
SystemError
(
myname
+
"(UDPReceiver): Unknown host. Use --udp-host"
);
port
=
conf
->
getArgInt
(
"--udp-port"
,
it
.
getProp
(
"port"
));
if
(
port
<=
0
)
throw
UniSetTypes
::
SystemError
(
myname
+
"(UDPReceiver): Unknown port address. Use --udp-port"
);
if
(
dlog
.
debugging
(
Debug
::
INFO
)
)
dlog
[
Debug
::
INFO
]
<<
"(UDPReceiver): UDP set to "
<<
s_host
<<
":"
<<
port
<<
endl
;
host
=
s_host
.
c_str
();
try
{
udp
=
new
ost
::
UDPDuplex
(
host
,
port
);
}
catch
(
ost
::
SockException
&
e
)
{
ostringstream
s
;
s
<<
e
.
getString
()
<<
": "
<<
e
.
getSystemErrorString
()
<<
endl
;
throw
SystemError
(
s
.
str
());
}
thr
=
new
ThreadCreator
<
UDPReceiver
>
(
this
,
&
UDPReceiver
::
poll
);
recvTimeout
=
conf
->
getArgPInt
(
"--udp-recv-timeout"
,
it
.
getProp
(
"recvTimeout"
),
5000
);
sendTimeout
=
conf
->
getArgPInt
(
"--udp-send-timeout"
,
it
.
getProp
(
"sendTimeout"
),
5000
);
polltime
=
conf
->
getArgPInt
(
"--udp-polltime"
,
it
.
getProp
(
"polltime"
),
100
);
// -------------------------------
// ********** HEARTBEAT *************
string
heart
=
conf
->
getArgParam
(
"--udp-heartbeat-id"
,
it
.
getProp
(
"heartbeat_id"
));
if
(
!
heart
.
empty
()
)
{
sidHeartBeat
=
conf
->
getSensorID
(
heart
);
if
(
sidHeartBeat
==
DefaultObjectId
)
{
ostringstream
err
;
err
<<
myname
<<
": не найден идентификатор для датчика 'HeartBeat' "
<<
heart
;
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(init): "
<<
err
.
str
()
<<
endl
;
throw
SystemError
(
err
.
str
());
}
int
heartbeatTime
=
getHeartBeatTime
();
if
(
heartbeatTime
)
ptHeartBeat
.
setTiming
(
heartbeatTime
);
else
ptHeartBeat
.
setTiming
(
UniSetTimer
::
WaitUpTime
);
maxHeartBeat
=
conf
->
getArgPInt
(
"--udp-heartbeat-max"
,
it
.
getProp
(
"heartbeat_max"
),
10
);
test_id
=
sidHeartBeat
;
}
else
{
test_id
=
conf
->
getSensorID
(
"TestMode_S"
);
if
(
test_id
==
DefaultObjectId
)
{
ostringstream
err
;
err
<<
myname
<<
"(init): test_id unknown. 'TestMode_S' not found..."
;
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(init): "
<<
err
.
str
()
<<
endl
;
throw
SystemError
(
err
.
str
());
}
}
dlog
[
Debug
::
INFO
]
<<
myname
<<
"(init): test_id="
<<
test_id
<<
endl
;
activateTimeout
=
conf
->
getArgPInt
(
"--activate-timeout"
,
20000
);
timeout_t
msec
=
conf
->
getArgPInt
(
"--udp-timeout"
,
it
.
getProp
(
"timeout"
),
3000
);
dlog
[
Debug
::
INFO
]
<<
myname
<<
"(init): udp-timeout="
<<
msec
<<
" msec"
<<
endl
;
}
// -----------------------------------------------------------------------------
UDPReceiver
::~
UDPReceiver
()
{
delete
udp
;
delete
shm
;
delete
thr
;
}
// -----------------------------------------------------------------------------
void
UDPReceiver
::
waitSMReady
()
{
// waiting for SM is ready...
int
ready_timeout
=
conf
->
getArgInt
(
"--udp-sm-ready-timeout"
,
"15000"
);
if
(
ready_timeout
==
0
)
ready_timeout
=
15000
;
else
if
(
ready_timeout
<
0
)
ready_timeout
=
UniSetTimer
::
WaitUpTime
;
if
(
!
shm
->
waitSMready
(
ready_timeout
,
50
)
)
{
ostringstream
err
;
err
<<
myname
<<
"(waitSMReady): Не дождались готовности SharedMemory к работе в течение "
<<
ready_timeout
<<
" мсек"
;
dlog
[
Debug
::
CRIT
]
<<
err
.
str
()
<<
endl
;
throw
SystemError
(
err
.
str
());
}
}
// -----------------------------------------------------------------------------
/*
void UDPReceiver::timerInfo( TimerMessage *tm )
{
if( tm->id == tmExchange )
step();
}
*/
// -----------------------------------------------------------------------------
void
UDPReceiver
::
step
()
{
// {
// uniset_mutex_lock l(pollMutex,2000);
// poll();
// }
if
(
!
activated
)
return
;
if
(
sidHeartBeat
!=
DefaultObjectId
&&
ptHeartBeat
.
checkTime
()
)
{
try
{
shm
->
localSaveValue
(
aitHeartBeat
,
sidHeartBeat
,
maxHeartBeat
,
getId
());
ptHeartBeat
.
reset
();
}
catch
(
Exception
&
ex
)
{
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(step): (hb) "
<<
ex
<<
std
::
endl
;
}
}
}
// -----------------------------------------------------------------------------
void
UDPReceiver
::
poll
()
{
try
{
// udp->connect(host,port);
// udp->UDPSocket::setPeer(host,port);
}
catch
(
UniSetTypes
::
Exception
&
ex
)
{
cerr
<<
myname
<<
"(step): "
<<
ex
<<
std
::
endl
;
// reise(SIGTERM);
return
;
}
while
(
activated
)
{
try
{
recv
();
// send();
}
catch
(
ost
::
SockException
&
e
)
{
cerr
<<
e
.
getString
()
<<
": "
<<
e
.
getSystemErrorString
()
<<
endl
;
}
catch
(
UniSetTypes
::
Exception
&
ex
)
{
cerr
<<
myname
<<
"(step): "
<<
ex
<<
std
::
endl
;
}
catch
(...)
{
cerr
<<
myname
<<
"(step): catch ..."
<<
std
::
endl
;
}
msleep
(
polltime
);
}
cerr
<<
"************* execute FINISH **********"
<<
endl
;
}
// -----------------------------------------------------------------------------
void
UDPReceiver
::
recv
()
{
cout
<<
myname
<<
": recv....(timeout="
<<
recvTimeout
<<
")"
<<
endl
;
UniSetUDP
::
UDPHeader
h
;
// receive
if
(
udp
->
isInputReady
(
recvTimeout
)
)
{
ssize_t
ret
=
udp
->
UDPReceive
::
receive
(
&
h
,
sizeof
(
h
));
if
(
ret
<
(
ssize_t
)
sizeof
(
h
)
)
{
cerr
<<
myname
<<
"(receive): ret="
<<
ret
<<
" sizeof="
<<
sizeof
(
h
)
<<
endl
;
return
;
}
cout
<<
myname
<<
"(receive): header: "
<<
h
<<
endl
;
if
(
h
.
dcount
<=
0
)
{
cout
<<
" data=0"
<<
endl
;
return
;
}
UniSetUDP
::
UDPData
d
;
// ignore echo...
#if 0
if( h.nodeID == conf->getLocalNode() && h.procID == getId() )
{
for( int i=0; i<h.dcount;i++ )
{
ssize_t ret = udp->UDPReceive::receive(&d,sizeof(d));
if( ret < (ssize_t)sizeof(d) )
return;
}
return;
}
#endif
for
(
int
i
=
0
;
i
<
h
.
dcount
;
i
++
)
{
ssize_t
ret
=
udp
->
UDPReceive
::
receive
(
&
d
,
sizeof
(
d
));
if
(
ret
<
(
ssize_t
)
sizeof
(
d
)
)
{
cerr
<<
myname
<<
"(receive data "
<<
i
<<
"): ret="
<<
ret
<<
" sizeof="
<<
sizeof
(
d
)
<<
endl
;
break
;
}
cout
<<
myname
<<
"(receive data "
<<
i
<<
"): "
<<
d
<<
endl
;
}
}
// else
// {
// cout << "no InputReady.." << endl;
// }
}
// -----------------------------------------------------------------------------
void
UDPReceiver
::
processingMessage
(
UniSetTypes
::
VoidMessage
*
msg
)
{
try
{
switch
(
msg
->
type
)
{
case
UniSetTypes
:
:
Message
::
SysCommand
:
{
UniSetTypes
::
SystemMessage
sm
(
msg
);
sysCommand
(
&
sm
);
}
break
;
case
Message
:
:
SensorInfo
:
{
SensorMessage
sm
(
msg
);
sensorInfo
(
&
sm
);
}
break
;
default
:
break
;
}
}
catch
(
SystemError
&
ex
)
{
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(SystemError): "
<<
ex
<<
std
::
endl
;
// throw SystemError(ex);
raise
(
SIGTERM
);
}
catch
(
Exception
&
ex
)
{
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(processingMessage): "
<<
ex
<<
std
::
endl
;
}
catch
(...)
{
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(processingMessage): catch ...
\n
"
;
}
}
// -----------------------------------------------------------------------------
void
UDPReceiver
::
sysCommand
(
UniSetTypes
::
SystemMessage
*
sm
)
{
switch
(
sm
->
command
)
{
case
SystemMessage
:
:
StartUp
:
{
waitSMReady
();
// подождать пока пройдёт инициализация датчиков
// см. activateObject()
msleep
(
initPause
);
PassiveTimer
ptAct
(
activateTimeout
);
while
(
!
activated
&&
!
ptAct
.
checkTime
()
)
{
cout
<<
myname
<<
"(sysCommand): wait activate..."
<<
endl
;
msleep
(
300
);
if
(
activated
)
break
;
}
if
(
!
activated
)
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(sysCommand): ************* don`t activate?! ************"
<<
endl
;
{
UniSetTypes
::
uniset_mutex_lock
l
(
mutex_start
,
10000
);
askSensors
(
UniversalIO
::
UIONotify
);
}
thr
->
start
();
}
case
SystemMessage
:
:
FoldUp
:
case
SystemMessage
:
:
Finish
:
askSensors
(
UniversalIO
::
UIODontNotify
);
break
;
case
SystemMessage
:
:
WatchDog
:
{
// ОПТИМИЗАЦИЯ (защита от двойного перезаказа при старте)
// Если идёт локальная работа
// (т.е. UDPReceiver запущен в одном процессе с SharedMemory2)
// то обрабатывать WatchDog не надо, т.к. мы и так ждём готовности SM
// при заказе датчиков, а если SM вылетит, то вместе с этим процессом(UDPReceiver)
if
(
shm
->
isLocalwork
()
)
break
;
askSensors
(
UniversalIO
::
UIONotify
);
}
break
;
case
SystemMessage
:
:
LogRotate
:
{
// переоткрываем логи
unideb
<<
myname
<<
"(sysCommand): logRotate"
<<
std
::
endl
;
string
fname
=
unideb
.
getLogFile
();
if
(
!
fname
.
empty
()
)
{
unideb
.
logFile
(
fname
);
unideb
<<
myname
<<
"(sysCommand): ***************** UNIDEB LOG ROTATE *****************"
<<
std
::
endl
;
}
dlog
<<
myname
<<
"(sysCommand): logRotate"
<<
std
::
endl
;
fname
=
dlog
.
getLogFile
();
if
(
!
fname
.
empty
()
)
{
dlog
.
logFile
(
fname
);
dlog
<<
myname
<<
"(sysCommand): ***************** dlog LOG ROTATE *****************"
<<
std
::
endl
;
}
}
break
;
default
:
break
;
}
}
// ------------------------------------------------------------------------------------------
void
UDPReceiver
::
askSensors
(
UniversalIO
::
UIOCommand
cmd
)
{
if
(
!
shm
->
waitSMworking
(
test_id
,
activateTimeout
,
50
)
)
{
ostringstream
err
;
err
<<
myname
<<
"(askSensors): Не дождались готовности(work) SharedMemory к работе в течение "
<<
activateTimeout
<<
" мсек"
;
dlog
[
Debug
::
CRIT
]
<<
err
.
str
()
<<
endl
;
kill
(
SIGTERM
,
getpid
());
// прерываем (перезапускаем) процесс...
throw
SystemError
(
err
.
str
());
}
}
// ------------------------------------------------------------------------------------------
void
UDPReceiver
::
sensorInfo
(
UniSetTypes
::
SensorMessage
*
sm
)
{
}
// ------------------------------------------------------------------------------------------
bool
UDPReceiver
::
activateObject
()
{
// блокирование обработки Starsp
// пока не пройдёт инициализация датчиков
// см. sysCommand()
{
activated
=
false
;
UniSetTypes
::
uniset_mutex_lock
l
(
mutex_start
,
5000
);
UniSetObject_LT
::
activateObject
();
initIterators
();
activated
=
true
;
}
return
true
;
}
// ------------------------------------------------------------------------------------------
void
UDPReceiver
::
sigterm
(
int
signo
)
{
cerr
<<
myname
<<
": ********* SIGTERM("
<<
signo
<<
") ********"
<<
endl
;
activated
=
false
;
udp
->
disconnect
();
UniSetObject_LT
::
sigterm
(
signo
);
}
// ------------------------------------------------------------------------------------------
void
UDPReceiver
::
initIterators
()
{
shm
->
initAIterator
(
aitHeartBeat
);
}
// -----------------------------------------------------------------------------
void
UDPReceiver
::
help_print
(
int
argc
,
char
*
argv
[]
)
{
cout
<<
"--udp-polltime msec - Пауза между опросаом карт. По умолчанию 200 мсек."
<<
endl
;
cout
<<
"--udp-heartbeat-id - Данный процесс связан с указанным аналоговым heartbeat-дачиком."
<<
endl
;
cout
<<
"--udp-heartbeat-max - Максимальное значение heartbeat-счётчика для данного процесса. По умолчанию 10."
<<
endl
;
cout
<<
"--udp-ready-timeout - Время ожидания готовности SM к работе, мсек. (-1 - ждать 'вечно')"
<<
endl
;
cout
<<
"--udp-initPause - Задержка перед инициализацией (время на активизация процесса)"
<<
endl
;
cout
<<
"--udp-notRespondSensor - датчик связи для данного процесса "
<<
endl
;
cout
<<
"--udp-sm-ready-timeout - время на ожидание старта SM"
<<
endl
;
cout
<<
" Настройки протокола RS: "
<<
endl
;
cout
<<
"--udp-dev devname - файл устройства"
<<
endl
;
cout
<<
"--udp-speed - Скорость обмена (9600,19920,38400,57600,115200)."
<<
endl
;
cout
<<
"--udp-my-addr - адрес текущего узла"
<<
endl
;
cout
<<
"--udp-recv-timeout - Таймаут на ожидание ответа."
<<
endl
;
}
// -----------------------------------------------------------------------------
UDPReceiver
*
UDPReceiver
::
init_udpreceiver
(
int
argc
,
char
*
argv
[],
UniSetTypes
::
ObjectId
icID
,
SharedMemory
*
ic
)
{
string
name
=
conf
->
getArgParam
(
"--udp-name"
,
"UDPReceiver1"
);
if
(
name
.
empty
()
)
{
cerr
<<
"(udpexchange): Не задан name'"
<<
endl
;
return
0
;
}
ObjectId
ID
=
conf
->
getObjectID
(
name
);
if
(
ID
==
UniSetTypes
::
DefaultObjectId
)
{
cerr
<<
"(udpexchange): идентификатор '"
<<
name
<<
"' не найден в конф. файле!"
<<
" в секции "
<<
conf
->
getObjectsSection
()
<<
endl
;
return
0
;
}
dlog
[
Debug
::
INFO
]
<<
"(rsexchange): name = "
<<
name
<<
"("
<<
ID
<<
")"
<<
endl
;
return
new
UDPReceiver
(
ID
,
icID
,
ic
);
}
// -----------------------------------------------------------------------------
std
::
ostream
&
operator
<<
(
std
::
ostream
&
os
,
UDPReceiver
::
UItem
&
p
)
{
return
os
<<
" sid="
<<
p
.
si
.
id
;
}
// -----------------------------------------------------------------------------
extensions/UDPExchange/UDPReceiver.h
0 → 100644
View file @
02375558
// $Id: UDPExchange.h,v 1.1 2009/02/10 20:38:27 vpashka Exp $
// -----------------------------------------------------------------------------
#ifndef UDPExchange_H_
#define UDPReceiver_H_
// -----------------------------------------------------------------------------
#include <ostream>
#include <string>
#include <vector>
#include <cc++/socket.h>
#include "UniSetObject_LT.h"
#include "Trigger.h"
#include "Mutex.h"
#include "SMInterface.h"
#include "SharedMemory.h"
#include "ThreadCreator.h"
#include "UDPPacket.h"
// -----------------------------------------------------------------------------
class
UDPReceiver
:
public
UniSetObject_LT
{
public
:
UDPReceiver
(
UniSetTypes
::
ObjectId
objId
,
UniSetTypes
::
ObjectId
shmID
,
SharedMemory
*
ic
=
0
);
virtual
~
UDPReceiver
();
/*! глобальная функция для инициализации объекта */
static
UDPReceiver
*
init_udpreceiver
(
int
argc
,
char
*
argv
[],
UniSetTypes
::
ObjectId
shmID
,
SharedMemory
*
ic
=
0
);
/*! глобальная функция для вывода help-а */
static
void
help_print
(
int
argc
,
char
*
argv
[]
);
struct
UItem
{
UItem
()
:
val
(
0
)
{}
IOController_i
::
SensorInfo
si
;
IOController
::
AIOStateList
::
iterator
ait
;
IOController
::
DIOStateList
::
iterator
dit
;
UniSetTypes
::
uniset_spin_mutex
val_lock
;
UniSetUDP
::
UDPMessage
::
UDPDataList
::
iterator
pack_it
;
long
val
;
friend
std
::
ostream
&
operator
<<
(
std
::
ostream
&
os
,
UItem
&
p
);
};
protected
:
xmlNode
*
cnode
;
std
::
string
s_field
;
std
::
string
s_fvalue
;
SMInterface
*
shm
;
void
poll
();
void
recv
();
void
step
();
virtual
void
processingMessage
(
UniSetTypes
::
VoidMessage
*
msg
);
void
sysCommand
(
UniSetTypes
::
SystemMessage
*
msg
);
void
sensorInfo
(
UniSetTypes
::
SensorMessage
*
sm
);
void
askSensors
(
UniversalIO
::
UIOCommand
cmd
);
void
waitSMReady
();
virtual
bool
activateObject
();
// действия при завершении работы
virtual
void
sigterm
(
int
signo
);
void
initIterators
();
private
:
UDPReceiver
();
bool
initPause
;
UniSetTypes
::
uniset_mutex
mutex_start
;
PassiveTimer
ptHeartBeat
;
UniSetTypes
::
ObjectId
sidHeartBeat
;
int
maxHeartBeat
;
IOController
::
AIOStateList
::
iterator
aitHeartBeat
;
UniSetTypes
::
ObjectId
test_id
;
int
polltime
;
/*!< переодичность обновления данных, [мсек] */
ost
::
UDPDuplex
*
udp
;
ost
::
IPV4Host
host
;
ost
::
tpport_t
port
;
UniSetTypes
::
uniset_mutex
pollMutex
;
Trigger
trTimeout
;
int
recvTimeout
;
int
sendTimeout
;
bool
activated
;
int
activateTimeout
;
ThreadCreator
<
UDPReceiver
>*
thr
;
};
// -----------------------------------------------------------------------------
#endif // UDPReceiver_H_
// -----------------------------------------------------------------------------
extensions/UDPExchange/UDPSender.cc
0 → 100644
View file @
02375558
#include <sstream>
#include "Exceptions.h"
#include "Extensions.h"
#include "UDPSender.h"
// -----------------------------------------------------------------------------
using
namespace
std
;
using
namespace
UniSetTypes
;
using
namespace
UniSetExtensions
;
// -----------------------------------------------------------------------------
UDPSender
::
UDPSender
(
UniSetTypes
::
ObjectId
objId
,
UniSetTypes
::
ObjectId
shmId
,
SharedMemory
*
ic
)
:
UniSetObject_LT
(
objId
),
shm
(
0
),
initPause
(
0
),
udp
(
0
),
activated
(
false
),
dlist
(
100
),
maxItem
(
0
)
{
if
(
objId
==
DefaultObjectId
)
throw
UniSetTypes
::
SystemError
(
"(UDPSender): objId=-1?!! Use --udp-name"
);
// xmlNode* cnode = conf->getNode(myname);
cnode
=
conf
->
getNode
(
myname
);
if
(
cnode
==
NULL
)
throw
UniSetTypes
::
SystemError
(
"(UDPSender): Not find conf-node for "
+
myname
);
shm
=
new
SMInterface
(
shmId
,
&
ui
,
objId
,
ic
);
UniXML_iterator
it
(
cnode
);
// определяем фильтр
s_field
=
conf
->
getArgParam
(
"--udp-filter-field"
);
s_fvalue
=
conf
->
getArgParam
(
"--udp-filter-value"
);
dlog
[
Debug
::
INFO
]
<<
myname
<<
"(init): read fileter-field='"
<<
s_field
<<
"' filter-value='"
<<
s_fvalue
<<
"'"
<<
endl
;
// ---------- init RS ----------
// UniXML_iterator it(cnode);
string
s_host
=
conf
->
getArgParam
(
"--udp-host"
,
it
.
getProp
(
"host"
));
if
(
s_host
.
empty
()
)
throw
UniSetTypes
::
SystemError
(
myname
+
"(UDPSender): Unknown host. Use --udp-host"
);
port
=
conf
->
getArgInt
(
"--udp-port"
,
it
.
getProp
(
"port"
));
if
(
port
<=
0
)
throw
UniSetTypes
::
SystemError
(
myname
+
"(UDPSender): Unknown port address. Use --udp-port"
);
bool
broadcast
=
conf
->
getArgInt
(
"--udp-broadcast"
,
it
.
getProp
(
"broadcast"
));
host
=
s_host
.
c_str
();
if
(
dlog
.
debugging
(
Debug
::
INFO
)
)
dlog
[
Debug
::
INFO
]
<<
"(UDPSender): UDP set to "
<<
s_host
<<
":"
<<
port
<<
" broadcast="
<<
broadcast
<<
endl
;
try
{
if
(
!
broadcast
)
udp
=
new
ost
::
UDPSocket
();
else
udp
=
new
ost
::
UDPBroadcast
(
host
,
port
);
}
catch
(
ost
::
SockException
&
e
)
{
ostringstream
s
;
s
<<
e
.
getString
()
<<
": "
<<
e
.
getSystemErrorString
();
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(init): "
<<
s
.
str
()
<<
endl
;
throw
SystemError
(
s
.
str
());
}
thr
=
new
ThreadCreator
<
UDPSender
>
(
this
,
&
UDPSender
::
poll
);
sendTimeout
=
conf
->
getArgPInt
(
"--udp-send-timeout"
,
it
.
getProp
(
"sendTimeout"
),
5000
);
sendtime
=
conf
->
getArgPInt
(
"--udp-sendtime"
,
it
.
getProp
(
"sendtime"
),
100
);
// -------------------------------
if
(
shm
->
isLocalwork
()
)
{
readConfiguration
();
dlist
.
resize
(
maxItem
);
dlog
[
Debug
::
INFO
]
<<
myname
<<
"(init): dlist size = "
<<
dlist
.
size
()
<<
endl
;
}
else
ic
->
addReadItem
(
sigc
::
mem_fun
(
this
,
&
UDPSender
::
readItem
)
);
// ********** HEARTBEAT *************
string
heart
=
conf
->
getArgParam
(
"--udp-heartbeat-id"
,
it
.
getProp
(
"heartbeat_id"
));
if
(
!
heart
.
empty
()
)
{
sidHeartBeat
=
conf
->
getSensorID
(
heart
);
if
(
sidHeartBeat
==
DefaultObjectId
)
{
ostringstream
err
;
err
<<
myname
<<
": не найден идентификатор для датчика 'HeartBeat' "
<<
heart
;
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(init): "
<<
err
.
str
()
<<
endl
;
throw
SystemError
(
err
.
str
());
}
int
heartbeatTime
=
getHeartBeatTime
();
if
(
heartbeatTime
)
ptHeartBeat
.
setTiming
(
heartbeatTime
);
else
ptHeartBeat
.
setTiming
(
UniSetTimer
::
WaitUpTime
);
maxHeartBeat
=
conf
->
getArgPInt
(
"--udp-heartbeat-max"
,
it
.
getProp
(
"heartbeat_max"
),
10
);
test_id
=
sidHeartBeat
;
}
else
{
test_id
=
conf
->
getSensorID
(
"TestMode_S"
);
if
(
test_id
==
DefaultObjectId
)
{
ostringstream
err
;
err
<<
myname
<<
"(init): test_id unknown. 'TestMode_S' not found..."
;
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(init): "
<<
err
.
str
()
<<
endl
;
throw
SystemError
(
err
.
str
());
}
}
dlog
[
Debug
::
INFO
]
<<
myname
<<
"(init): test_id="
<<
test_id
<<
endl
;
activateTimeout
=
conf
->
getArgPInt
(
"--activate-timeout"
,
20000
);
timeout_t
msec
=
conf
->
getArgPInt
(
"--udp-timeout"
,
it
.
getProp
(
"timeout"
),
3000
);
dlog
[
Debug
::
INFO
]
<<
myname
<<
"(init): udp-timeout="
<<
msec
<<
" msec"
<<
endl
;
}
// -----------------------------------------------------------------------------
UDPSender
::~
UDPSender
()
{
delete
udp
;
delete
shm
;
delete
thr
;
}
// -----------------------------------------------------------------------------
void
UDPSender
::
waitSMReady
()
{
// waiting for SM is ready...
int
ready_timeout
=
conf
->
getArgInt
(
"--udp-sm-ready-timeout"
,
"15000"
);
if
(
ready_timeout
==
0
)
ready_timeout
=
15000
;
else
if
(
ready_timeout
<
0
)
ready_timeout
=
UniSetTimer
::
WaitUpTime
;
if
(
!
shm
->
waitSMready
(
ready_timeout
,
50
)
)
{
ostringstream
err
;
err
<<
myname
<<
"(waitSMReady): Не дождались готовности SharedMemory к работе в течение "
<<
ready_timeout
<<
" мсек"
;
dlog
[
Debug
::
CRIT
]
<<
err
.
str
()
<<
endl
;
throw
SystemError
(
err
.
str
());
}
}
// -----------------------------------------------------------------------------
void
UDPSender
::
step
()
{
if
(
!
activated
)
return
;
if
(
sidHeartBeat
!=
DefaultObjectId
&&
ptHeartBeat
.
checkTime
()
)
{
try
{
shm
->
localSaveValue
(
aitHeartBeat
,
sidHeartBeat
,
maxHeartBeat
,
getId
());
ptHeartBeat
.
reset
();
}
catch
(
Exception
&
ex
)
{
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(step): (hb) "
<<
ex
<<
std
::
endl
;
}
}
}
// -----------------------------------------------------------------------------
void
UDPSender
::
poll
()
{
dlist
.
resize
(
maxItem
);
dlog
[
Debug
::
INFO
]
<<
myname
<<
"(init): dlist size = "
<<
dlist
.
size
()
<<
endl
;
try
{
udp
->
setPeer
(
host
,
port
);
}
catch
(
ost
::
SockException
&
e
)
{
ostringstream
s
;
s
<<
e
.
getString
()
<<
": "
<<
e
.
getSystemErrorString
();
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(poll): "
<<
s
.
str
()
<<
endl
;
throw
SystemError
(
s
.
str
());
}
while
(
activated
)
{
try
{
send
();
}
catch
(
ost
::
SockException
&
e
)
{
cerr
<<
e
.
getString
()
<<
": "
<<
e
.
getSystemErrorString
()
<<
endl
;
}
catch
(
UniSetTypes
::
Exception
&
ex
)
{
cerr
<<
myname
<<
"(step): "
<<
ex
<<
std
::
endl
;
}
catch
(...)
{
cerr
<<
myname
<<
"(step): catch ..."
<<
std
::
endl
;
}
msleep
(
sendtime
);
}
cerr
<<
"************* execute FINISH **********"
<<
endl
;
}
// -----------------------------------------------------------------------------
void
UDPSender
::
send
()
{
cout
<<
myname
<<
": send..."
<<
endl
;
UniSetUDP
::
UDPHeader
h
;
h
.
nodeID
=
conf
->
getLocalNode
();
h
.
procID
=
getId
();
h
.
dcount
=
mypack
.
size
();
// receive
ssize_t
ret
=
udp
->
send
((
char
*
)(
&
h
),
sizeof
(
h
));
if
(
ret
<
(
ssize_t
)
sizeof
(
h
)
)
{
cerr
<<
myname
<<
"(send data header): ret="
<<
ret
<<
" sizeof="
<<
sizeof
(
h
)
<<
endl
;
return
;
}
#warning use mutex for list!!!
UniSetUDP
::
UDPMessage
::
UDPDataList
::
iterator
it
=
mypack
.
dlist
.
begin
();
for
(
;
it
!=
mypack
.
dlist
.
end
();
++
it
)
{
cout
<<
myname
<<
"(send): "
<<
(
*
it
)
<<
endl
;
ssize_t
ret
=
udp
->
send
((
char
*
)(
&
(
*
it
)),
sizeof
(
*
it
));
if
(
ret
<
(
ssize_t
)
sizeof
(
*
it
)
)
{
cerr
<<
myname
<<
"(send data): ret="
<<
ret
<<
" sizeof="
<<
sizeof
(
*
it
)
<<
endl
;
break
;
}
}
}
// -----------------------------------------------------------------------------
void
UDPSender
::
processingMessage
(
UniSetTypes
::
VoidMessage
*
msg
)
{
try
{
switch
(
msg
->
type
)
{
case
UniSetTypes
:
:
Message
::
SysCommand
:
{
UniSetTypes
::
SystemMessage
sm
(
msg
);
sysCommand
(
&
sm
);
}
break
;
case
Message
:
:
SensorInfo
:
{
SensorMessage
sm
(
msg
);
sensorInfo
(
&
sm
);
}
break
;
default
:
break
;
}
}
catch
(
SystemError
&
ex
)
{
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(SystemError): "
<<
ex
<<
std
::
endl
;
// throw SystemError(ex);
raise
(
SIGTERM
);
}
catch
(
Exception
&
ex
)
{
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(processingMessage): "
<<
ex
<<
std
::
endl
;
}
catch
(...)
{
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(processingMessage): catch ...
\n
"
;
}
}
// -----------------------------------------------------------------------------
void
UDPSender
::
sysCommand
(
UniSetTypes
::
SystemMessage
*
sm
)
{
switch
(
sm
->
command
)
{
case
SystemMessage
:
:
StartUp
:
{
waitSMReady
();
// подождать пока пройдёт инициализация датчиков
// см. activateObject()
msleep
(
initPause
);
PassiveTimer
ptAct
(
activateTimeout
);
while
(
!
activated
&&
!
ptAct
.
checkTime
()
)
{
cout
<<
myname
<<
"(sysCommand): wait activate..."
<<
endl
;
msleep
(
300
);
if
(
activated
)
break
;
}
if
(
!
activated
)
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(sysCommand): ************* don`t activate?! ************"
<<
endl
;
{
UniSetTypes
::
uniset_mutex_lock
l
(
mutex_start
,
10000
);
askSensors
(
UniversalIO
::
UIONotify
);
}
thr
->
start
();
}
case
SystemMessage
:
:
FoldUp
:
case
SystemMessage
:
:
Finish
:
askSensors
(
UniversalIO
::
UIODontNotify
);
break
;
case
SystemMessage
:
:
WatchDog
:
{
// ОПТИМИЗАЦИЯ (защита от двойного перезаказа при старте)
// Если идёт локальная работа
// (т.е. UDPSender запущен в одном процессе с SharedMemory2)
// то обрабатывать WatchDog не надо, т.к. мы и так ждём готовности SM
// при заказе датчиков, а если SM вылетит, то вместе с этим процессом(UDPSender)
if
(
shm
->
isLocalwork
()
)
break
;
askSensors
(
UniversalIO
::
UIONotify
);
}
break
;
case
SystemMessage
:
:
LogRotate
:
{
// переоткрываем логи
unideb
<<
myname
<<
"(sysCommand): logRotate"
<<
std
::
endl
;
string
fname
=
unideb
.
getLogFile
();
if
(
!
fname
.
empty
()
)
{
unideb
.
logFile
(
fname
);
unideb
<<
myname
<<
"(sysCommand): ***************** UNIDEB LOG ROTATE *****************"
<<
std
::
endl
;
}
dlog
<<
myname
<<
"(sysCommand): logRotate"
<<
std
::
endl
;
fname
=
dlog
.
getLogFile
();
if
(
!
fname
.
empty
()
)
{
dlog
.
logFile
(
fname
);
dlog
<<
myname
<<
"(sysCommand): ***************** dlog LOG ROTATE *****************"
<<
std
::
endl
;
}
}
break
;
default
:
break
;
}
}
// ------------------------------------------------------------------------------------------
void
UDPSender
::
askSensors
(
UniversalIO
::
UIOCommand
cmd
)
{
if
(
!
shm
->
waitSMworking
(
test_id
,
activateTimeout
,
50
)
)
{
ostringstream
err
;
err
<<
myname
<<
"(askSensors): Не дождались готовности(work) SharedMemory к работе в течение "
<<
activateTimeout
<<
" мсек"
;
dlog
[
Debug
::
CRIT
]
<<
err
.
str
()
<<
endl
;
kill
(
SIGTERM
,
getpid
());
// прерываем (перезапускаем) процесс...
throw
SystemError
(
err
.
str
());
}
DMap
::
iterator
it
=
dlist
.
begin
();
for
(
;
it
!=
dlist
.
end
();
++
it
)
{
try
{
shm
->
askSensor
(
it
->
si
.
id
,
cmd
);
}
catch
(
UniSetTypes
::
Exception
&
ex
)
{
dlog
[
Debug
::
WARN
]
<<
myname
<<
"(askSensors): "
<<
ex
<<
std
::
endl
;
}
catch
(...){}
}
}
// ------------------------------------------------------------------------------------------
void
UDPSender
::
sensorInfo
(
UniSetTypes
::
SensorMessage
*
sm
)
{
DMap
::
iterator
it
=
dlist
.
begin
();
for
(
;
it
!=
dlist
.
end
();
++
it
)
{
if
(
it
->
si
.
id
==
sm
->
id
)
{
uniset_spin_lock
lock
(
it
->
val_lock
);
it
->
val
=
sm
->
value
;
if
(
it
->
pack_it
!=
mypack
.
dlist
.
end
()
)
it
->
pack_it
->
val
=
sm
->
value
;
}
break
;
}
}
// ------------------------------------------------------------------------------------------
bool
UDPSender
::
activateObject
()
{
// блокирование обработки StarUp
// пока не пройдёт инициализация датчиков
// см. sysCommand()
{
activated
=
false
;
UniSetTypes
::
uniset_mutex_lock
l
(
mutex_start
,
5000
);
UniSetObject_LT
::
activateObject
();
initIterators
();
activated
=
true
;
}
return
true
;
}
// ------------------------------------------------------------------------------------------
void
UDPSender
::
sigterm
(
int
signo
)
{
cerr
<<
myname
<<
": ********* SIGTERM("
<<
signo
<<
") ********"
<<
endl
;
activated
=
false
;
udp
->
disconnect
();
UniSetObject_LT
::
sigterm
(
signo
);
}
// ------------------------------------------------------------------------------------------
void
UDPSender
::
readConfiguration
()
{
#warning Сделать сортировку по диапазонам адресов!!!
// чтобы запрашивать одним запросом, сразу несколько входов...
// readconf_ok = false;
xmlNode
*
root
=
conf
->
getXMLSensorsSection
();
if
(
!
root
)
{
ostringstream
err
;
err
<<
myname
<<
"(readConfiguration): не нашли корневого раздела <sensors>"
;
throw
SystemError
(
err
.
str
());
}
UniXML_iterator
it
(
root
);
if
(
!
it
.
goChildren
()
)
{
std
::
cerr
<<
myname
<<
"(readConfiguration): раздел <sensors> не содержит секций ?!!
\n
"
;
return
;
}
for
(
;
it
.
getCurrent
();
it
.
goNext
()
)
{
if
(
check_item
(
it
)
)
initItem
(
it
);
}
// readconf_ok = true;
}
// ------------------------------------------------------------------------------------------
bool
UDPSender
::
check_item
(
UniXML_iterator
&
it
)
{
if
(
s_field
.
empty
()
)
return
true
;
// просто проверка на не пустой field
if
(
s_fvalue
.
empty
()
&&
it
.
getProp
(
s_field
).
empty
()
)
return
false
;
// просто проверка что field = value
if
(
!
s_fvalue
.
empty
()
&&
it
.
getProp
(
s_field
)
!=
s_fvalue
)
return
false
;
return
true
;
}
// ------------------------------------------------------------------------------------------
bool
UDPSender
::
readItem
(
UniXML
&
xml
,
UniXML_iterator
&
it
,
xmlNode
*
sec
)
{
if
(
check_item
(
it
)
)
initItem
(
it
);
return
true
;
}
// ------------------------------------------------------------------------------------------
bool
UDPSender
::
initItem
(
UniXML_iterator
&
it
)
{
string
sname
(
it
.
getProp
(
"name"
)
);
string
tid
=
it
.
getProp
(
"id"
);
ObjectId
sid
;
if
(
!
tid
.
empty
()
)
{
sid
=
UniSetTypes
::
uni_atoi
(
tid
);
if
(
sid
<=
0
)
sid
=
DefaultObjectId
;
}
else
sid
=
conf
->
getSensorID
(
sname
);
if
(
sid
==
DefaultObjectId
)
{
if
(
dlog
)
dlog
[
Debug
::
CRIT
]
<<
myname
<<
"(readItem): ID not found for "
<<
sname
<<
endl
;
return
false
;
}
UItem
p
;
p
.
si
.
id
=
sid
;
p
.
si
.
node
=
conf
->
getLocalNode
();
mypack
.
addData
(
sid
,
0
);
p
.
pack_it
=
(
mypack
.
dlist
.
end
()
--
);
if
(
maxItem
>=
dlist
.
size
()
)
dlist
.
resize
(
maxItem
+
10
);
dlist
[
maxItem
]
=
p
;
maxItem
++
;
if
(
dlog
.
debugging
(
Debug
::
INFO
)
)
dlog
[
Debug
::
INFO
]
<<
myname
<<
"(initItem): add "
<<
p
<<
endl
;
return
true
;
}
// ------------------------------------------------------------------------------------------
void
UDPSender
::
initIterators
()
{
DMap
::
iterator
it
=
dlist
.
begin
();
for
(
;
it
!=
dlist
.
end
();
it
++
)
{
shm
->
initDIterator
(
it
->
dit
);
shm
->
initAIterator
(
it
->
ait
);
}
shm
->
initAIterator
(
aitHeartBeat
);
}
// -----------------------------------------------------------------------------
void
UDPSender
::
help_print
(
int
argc
,
char
*
argv
[]
)
{
cout
<<
"--udp-sendtime msec - Пауза между опросами. По умолчанию 200 мсек."
<<
endl
;
cout
<<
"--udp-heartbeat-id - Данный процесс связан с указанным аналоговым heartbeat-дачиком."
<<
endl
;
cout
<<
"--udp-heartbeat-max - Максимальное значение heartbeat-счётчика для данного процесса. По умолчанию 10."
<<
endl
;
cout
<<
"--udp-ready-timeout - Время ожидания готовности SM к работе, мсек. (-1 - ждать 'вечно')"
<<
endl
;
cout
<<
"--udp-initPause - Задержка перед инициализацией (время на активизация процесса)"
<<
endl
;
cout
<<
"--udp-sm-ready-timeout - время на ожидание старта SM"
<<
endl
;
cout
<<
" Настройки протокола UDP: "
<<
endl
;
cout
<<
"--udp-host [ip|hostname] - Адрес сервера"
<<
endl
;
cout
<<
"--udp-port - Порт."
<<
endl
;
cout
<<
"--udp-send-timeout - Таймаут на посылку ответа."
<<
endl
;
}
// -----------------------------------------------------------------------------
UDPSender
*
UDPSender
::
init_udpsender
(
int
argc
,
char
*
argv
[],
UniSetTypes
::
ObjectId
icID
,
SharedMemory
*
ic
)
{
string
name
=
conf
->
getArgParam
(
"--udp-name"
,
"UDPSender1"
);
if
(
name
.
empty
()
)
{
cerr
<<
"(UDPSender): Не задан name'"
<<
endl
;
return
0
;
}
ObjectId
ID
=
conf
->
getObjectID
(
name
);
if
(
ID
==
UniSetTypes
::
DefaultObjectId
)
{
cerr
<<
"(UDPSender): идентификатор '"
<<
name
<<
"' не найден в конф. файле!"
<<
" в секции "
<<
conf
->
getObjectsSection
()
<<
endl
;
return
0
;
}
dlog
[
Debug
::
INFO
]
<<
"(rsexchange): name = "
<<
name
<<
"("
<<
ID
<<
")"
<<
endl
;
return
new
UDPSender
(
ID
,
icID
,
ic
);
}
// -----------------------------------------------------------------------------
std
::
ostream
&
operator
<<
(
std
::
ostream
&
os
,
UDPSender
::
UItem
&
p
)
{
return
os
<<
" sid="
<<
p
.
si
.
id
;
}
// -----------------------------------------------------------------------------
extensions/UDPExchange/UDPSender.h
0 → 100644
View file @
02375558
#ifndef UDPSender_H_
#define UDPSender_H_
// -----------------------------------------------------------------------------
#include <ostream>
#include <string>
#include <vector>
#include <cc++/socket.h>
#include "UniSetObject_LT.h"
#include "Trigger.h"
#include "Mutex.h"
#include "SMInterface.h"
#include "SharedMemory.h"
#include "ThreadCreator.h"
#include "UDPPacket.h"
// -----------------------------------------------------------------------------
class
UDPSender
:
public
UniSetObject_LT
{
public
:
UDPSender
(
UniSetTypes
::
ObjectId
objId
,
UniSetTypes
::
ObjectId
shmID
,
SharedMemory
*
ic
=
0
);
virtual
~
UDPSender
();
/*! глобальная функция для инициализации объекта */
static
UDPSender
*
init_udpsender
(
int
argc
,
char
*
argv
[],
UniSetTypes
::
ObjectId
shmID
,
SharedMemory
*
ic
=
0
);
/*! глобальная функция для вывода help-а */
static
void
help_print
(
int
argc
,
char
*
argv
[]
);
struct
UItem
{
UItem
()
:
val
(
0
)
{}
IOController_i
::
SensorInfo
si
;
IOController
::
AIOStateList
::
iterator
ait
;
IOController
::
DIOStateList
::
iterator
dit
;
UniSetTypes
::
uniset_spin_mutex
val_lock
;
UniSetUDP
::
UDPMessage
::
UDPDataList
::
iterator
pack_it
;
long
val
;
friend
std
::
ostream
&
operator
<<
(
std
::
ostream
&
os
,
UItem
&
p
);
};
protected
:
xmlNode
*
cnode
;
std
::
string
s_field
;
std
::
string
s_fvalue
;
SMInterface
*
shm
;
void
poll
();
void
recv
();
void
send
();
void
step
();
virtual
void
processingMessage
(
UniSetTypes
::
VoidMessage
*
msg
);
void
sysCommand
(
UniSetTypes
::
SystemMessage
*
msg
);
void
sensorInfo
(
UniSetTypes
::
SensorMessage
*
sm
);
void
askSensors
(
UniversalIO
::
UIOCommand
cmd
);
void
waitSMReady
();
virtual
bool
activateObject
();
// действия при завершении работы
virtual
void
sigterm
(
int
signo
);
void
initIterators
();
bool
initItem
(
UniXML_iterator
&
it
);
bool
readItem
(
UniXML
&
xml
,
UniXML_iterator
&
it
,
xmlNode
*
sec
);
void
readConfiguration
();
bool
check_item
(
UniXML_iterator
&
it
);
private
:
UDPSender
();
bool
initPause
;
UniSetTypes
::
uniset_mutex
mutex_start
;
PassiveTimer
ptHeartBeat
;
UniSetTypes
::
ObjectId
sidHeartBeat
;
int
maxHeartBeat
;
IOController
::
AIOStateList
::
iterator
aitHeartBeat
;
UniSetTypes
::
ObjectId
test_id
;
int
sendtime
;
/*!< переодичность посылки данных, [мсек] */
ost
::
UDPSocket
*
udp
;
ost
::
IPV4Host
host
;
ost
::
tpport_t
port
;
UniSetTypes
::
uniset_mutex
sendMutex
;
Trigger
trTimeout
;
int
sendTimeout
;
bool
activated
;
int
activateTimeout
;
UniSetUDP
::
UDPMessage
mypack
;
typedef
std
::
vector
<
UItem
>
DMap
;
DMap
dlist
;
int
maxItem
;
ThreadCreator
<
UDPSender
>*
thr
;
};
// -----------------------------------------------------------------------------
#endif // UDPSender_H_
// -----------------------------------------------------------------------------
extensions/UDPExchange/start_fg.sh
View file @
02375558
#!/bin/sh
#!/bin/sh
uniset-start.sh
-f
./uniset-udpexchange
--udp-name
UDPExchange
--udp-host
localhost
--udp-port
2049
\
uniset-start.sh
-f
./uniset-udpexchange
--udp-name
UDPExchange
--udp-host
192.168.56.255
--udp-port
2050
--udp-broadcast
1
\
--confile
test.xml
\
--confile
test.xml
\
--udp-filter-field
udp
--udp-filter-value
1
\
--udp-filter-field
udp
--udp-filter-value
1
\
--udp-ip
\
--dlog-add-levels
info,crit,warn
--dlog-add-levels
info,crit,warn
extensions/UDPExchange/start_fg_receiver.sh
0 → 100755
View file @
02375558
#!/bin/sh
uniset-start.sh
-f
./uniset-udpreceiver
--udp-name
UDPExchange
\
--udp-host
192.168.56.255
--udp-port
2050
\
--confile
test.xml
\
--udp-filter-field
udp
--udp-filter-value
1
\
--dlog-add-levels
info,crit,warn
extensions/UDPExchange/start_fg_sender.sh
0 → 100755
View file @
02375558
#!/bin/sh
uniset-start.sh
-f
./uniset-udpsender
--udp-name
UDPExchange
\
--udp-host
192.168.56.255
--udp-port
2050
--udp-broadcast
1
\
--udp-sendtime
100
\
--confile
test.xml
\
--dlog-add-levels
info,crit,warn
# --udp-filter-field udp --udp-filter-value 1 \
extensions/UDPExchange/udpexchange.cc
View file @
02375558
...
@@ -11,6 +11,46 @@ using namespace UniSetExtensions;
...
@@ -11,6 +11,46 @@ using namespace UniSetExtensions;
// -----------------------------------------------------------------------------
// -----------------------------------------------------------------------------
int
main
(
int
argc
,
char
**
argv
)
int
main
(
int
argc
,
char
**
argv
)
{
{
ost
::
IPV4Host
host
=
"192.168.56.10"
;
ost
::
tpport_t
port
=
2050
;
try
{
ost
::
UDPSocket
*
udp
=
new
ost
::
UDPSocket
();
udp
->
setPeer
(
host
,
port
);
// udp->UDPTransmit::setBroadcast(false);
// udp->UDPTransmit::setRouting(false);
UniSetUDP
::
UDPHeader
h
;
h
.
nodeID
=
20
;
h
.
procID
=
20
;
h
.
dcount
=
0
;
// udp->connect(host,port);
// ost::IPV4Host h1 = udp->getPeer();
// cout << "peer: " << h1 << endl;
//if( udp->isOutputReady(5000) )
while
(
1
)
{
ssize_t
ret
=
udp
->
send
((
char
*
)(
&
h
),
sizeof
(
h
));
if
(
ret
<
(
ssize_t
)
sizeof
(
h
)
)
{
cerr
<<
"(send data header): ret="
<<
ret
<<
" sizeof="
<<
sizeof
(
h
)
<<
endl
;
// return 0;
}
else
cout
<<
"SEND OK..."
<<
endl
;
msleep
(
1000
);
}
}
catch
(
ost
::
SockException
&
e
)
{
cerr
<<
e
.
getString
()
<<
": "
<<
e
.
getSystemErrorString
()
<<
endl
;
return
0
;
}
try
try
{
{
if
(
argc
>
1
&&
(
!
strcmp
(
argv
[
1
],
"--help"
)
||
!
strcmp
(
argv
[
1
],
"-h"
))
)
if
(
argc
>
1
&&
(
!
strcmp
(
argv
[
1
],
"--help"
)
||
!
strcmp
(
argv
[
1
],
"-h"
))
)
...
...
extensions/UDPExchange/udpreceiver.cc
0 → 100644
View file @
02375558
// $Id: udpexchange.cc,v 1.1 2009/02/10 20:38:27 vpashka Exp $
// -----------------------------------------------------------------------------
#include <sstream>
#include "ObjectsActivator.h"
#include "Extensions.h"
#include "UDPReceiver.h"
// -----------------------------------------------------------------------------
using
namespace
std
;
using
namespace
UniSetTypes
;
using
namespace
UniSetExtensions
;
// -----------------------------------------------------------------------------
int
main
(
int
argc
,
char
**
argv
)
{
try
{
if
(
argc
>
1
&&
(
!
strcmp
(
argv
[
1
],
"--help"
)
||
!
strcmp
(
argv
[
1
],
"-h"
))
)
{
cout
<<
"--smemory-id objectName - SharedMemory objectID. Default: read from <SharedMemory>"
<<
endl
;
cout
<<
"--confile filename - configuration file. Default: configure.xml"
<<
endl
;
cout
<<
"--udp-logfile filename - logfilename. Default: udpexchange.log"
<<
endl
;
cout
<<
endl
;
UDPReceiver
::
help_print
(
argc
,
argv
);
return
0
;
}
string
confile
=
UniSetTypes
::
getArgParam
(
"--confile"
,
argc
,
argv
,
"configure.xml"
);
conf
=
new
Configuration
(
argc
,
argv
,
confile
);
string
logfilename
(
conf
->
getArgParam
(
"--udp-logfile"
));
if
(
logfilename
.
empty
()
)
logfilename
=
"udpexchange.log"
;
conf
->
initDebug
(
dlog
,
"dlog"
);
std
::
ostringstream
logname
;
string
dir
(
conf
->
getLogDir
());
logname
<<
dir
<<
logfilename
;
unideb
.
logFile
(
logname
.
str
()
);
dlog
.
logFile
(
logname
.
str
()
);
ObjectId
shmID
=
DefaultObjectId
;
string
sID
=
conf
->
getArgParam
(
"--smemory-id"
);
if
(
!
sID
.
empty
()
)
shmID
=
conf
->
getControllerID
(
sID
);
else
shmID
=
getSharedMemoryID
();
if
(
shmID
==
DefaultObjectId
)
{
cerr
<<
sID
<<
"? SharedMemoryID not found in "
<<
conf
->
getControllersSection
()
<<
" section"
<<
endl
;
return
1
;
}
UDPReceiver
*
udp
=
UDPReceiver
::
init_udpreceiver
(
argc
,
argv
,
shmID
);
if
(
!
udp
)
{
dlog
[
Debug
::
CRIT
]
<<
"(udpreceiver): init не прошёл..."
<<
endl
;
return
1
;
}
ObjectsActivator
act
;
act
.
addObject
(
static_cast
<
class
UniSetObject
*>
(
udp
));
SystemMessage
sm
(
SystemMessage
::
StartUp
);
act
.
broadcast
(
sm
.
transport_msg
()
);
unideb
(
Debug
::
ANY
)
<<
"
\n\n\n
"
;
unideb
[
Debug
::
ANY
]
<<
"(main): -------------- UDPRecevier START -------------------------
\n\n
"
;
dlog
(
Debug
::
ANY
)
<<
"
\n\n\n
"
;
dlog
[
Debug
::
ANY
]
<<
"(main): -------------- UDPReceiver START -------------------------
\n\n
"
;
act
.
run
(
false
);
// msleep(500);
// rs->execute();
}
catch
(
Exception
&
ex
)
{
dlog
[
Debug
::
CRIT
]
<<
"(udpexchange): "
<<
ex
<<
std
::
endl
;
}
catch
(
ost
::
SockException
&
e
)
{
ostringstream
s
;
s
<<
e
.
getString
()
<<
": "
<<
e
.
getSystemErrorString
();
dlog
[
Debug
::
CRIT
]
<<
s
.
str
()
<<
endl
;
}
catch
(...)
{
dlog
[
Debug
::
CRIT
]
<<
"(udpexchange): catch ..."
<<
std
::
endl
;
}
return
0
;
}
extensions/UDPExchange/udpsender.cc
0 → 100644
View file @
02375558
#include <sstream>
#include "ObjectsActivator.h"
#include "Extensions.h"
#include "UDPSender.h"
// -----------------------------------------------------------------------------
using
namespace
std
;
using
namespace
UniSetTypes
;
using
namespace
UniSetExtensions
;
// -----------------------------------------------------------------------------
int
main
(
int
argc
,
char
**
argv
)
{
try
{
if
(
argc
>
1
&&
(
!
strcmp
(
argv
[
1
],
"--help"
)
||
!
strcmp
(
argv
[
1
],
"-h"
))
)
{
cout
<<
"--smemory-id objectName - SharedMemory objectID. Default: read from <SharedMemory>"
<<
endl
;
cout
<<
"--confile filename - configuration file. Default: configure.xml"
<<
endl
;
cout
<<
"--udp-logfile filename - logfilename. Default: udpexchange.log"
<<
endl
;
cout
<<
endl
;
UDPSender
::
help_print
(
argc
,
argv
);
return
0
;
}
string
confile
=
UniSetTypes
::
getArgParam
(
"--confile"
,
argc
,
argv
,
"configure.xml"
);
conf
=
new
Configuration
(
argc
,
argv
,
confile
);
string
logfilename
(
conf
->
getArgParam
(
"--udp-logfile"
));
if
(
logfilename
.
empty
()
)
logfilename
=
"udpexchange.log"
;
conf
->
initDebug
(
dlog
,
"dlog"
);
std
::
ostringstream
logname
;
string
dir
(
conf
->
getLogDir
());
logname
<<
dir
<<
logfilename
;
unideb
.
logFile
(
logname
.
str
()
);
dlog
.
logFile
(
logname
.
str
()
);
ObjectId
shmID
=
DefaultObjectId
;
string
sID
=
conf
->
getArgParam
(
"--smemory-id"
);
if
(
!
sID
.
empty
()
)
shmID
=
conf
->
getControllerID
(
sID
);
else
shmID
=
getSharedMemoryID
();
if
(
shmID
==
DefaultObjectId
)
{
cerr
<<
sID
<<
"? SharedMemoryID not found in "
<<
conf
->
getControllersSection
()
<<
" section"
<<
endl
;
return
1
;
}
UDPSender
*
udp
=
UDPSender
::
init_udpsender
(
argc
,
argv
,
shmID
);
if
(
!
udp
)
{
dlog
[
Debug
::
CRIT
]
<<
"(udpsender): init не прошёл..."
<<
endl
;
return
1
;
}
ObjectsActivator
act
;
act
.
addObject
(
static_cast
<
class
UniSetObject
*>
(
udp
));
SystemMessage
sm
(
SystemMessage
::
StartUp
);
act
.
broadcast
(
sm
.
transport_msg
()
);
unideb
(
Debug
::
ANY
)
<<
"
\n\n\n
"
;
unideb
[
Debug
::
ANY
]
<<
"(main): -------------- UDPSender START -------------------------
\n\n
"
;
dlog
(
Debug
::
ANY
)
<<
"
\n\n\n
"
;
dlog
[
Debug
::
ANY
]
<<
"(main): -------------- UDPSender START -------------------------
\n\n
"
;
act
.
run
(
false
);
// msleep(500);
// rs->execute();
}
catch
(
Exception
&
ex
)
{
dlog
[
Debug
::
CRIT
]
<<
"(udpsender): "
<<
ex
<<
std
::
endl
;
}
catch
(
ost
::
SockException
&
e
)
{
dlog
[
Debug
::
CRIT
]
<<
"(udpsender): "
<<
e
.
getSystemErrorString
()
<<
endl
;
}
catch
(...)
{
dlog
[
Debug
::
CRIT
]
<<
"(udpsender): catch ..."
<<
std
::
endl
;
}
return
0
;
}
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment