--- /dev/null
+# vim: ts=4 sw=4 noet:
+
+# -------------------------------------------------------------------------------
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# -------------------------------------------------------------------------------
+
+# CAUTION: This file eventually should exist in the ci directory, however until
+# this can be confirmed, and the .yaml file(s) in the ci project changed
+# to indicaate that ci/Dockerfile should be used, this is here with minor
+# changes needed to exist at the root.
+# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+# CI to verify the MC application components
+# Inherits C toolchain from buildpack-deps:stretch then adds cmake and better shell(s).
+
+# It is assumed that this docker file is used with a build command run at the
+# root level of the repo (directory containing the ci directory). E.g.
+# docker build -f ci/Dockerfile .
+
+FROM buildpack-deps:stretch
+
+RUN apt-get update && apt-get -q -y install cmake ksh
+
+# stuff our repo things into a scratch area
+RUN mkdir /playpen
+ADD . /playpen
+
+
+# add any unit test scripts that need to be driven, e.g.
+# RUN ksh test/mcl_unit_test.ksh
+
+# This is a final/only script which might print useful things to the log and ALWAYS succeeds.
+RUN ksh /playpen/ci/stats.ksh
+
+# there is no cmd; the build/verification assumes that if the image is created
+# successfully, e.g. none of the previous run commands fail, that the environment
+# is successfully vetted.
--- /dev/null
+
+ Unless otherwise specified, all software contained herein is licensed
+ under the Apache License, Version 2.0 (the "Software License");
+ you may not use this software except in compliance with the Software
+ License. You may obtain a copy of the Software License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the Software License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the Software License for the specific language governing permissions
+ and limitations under the Software License.
+
+
+
+ Unless otherwise specified, all documentation contained herein is licensed
+ under the Creative Commons License, Attribution 4.0 Intl. (the
+ "Documentation License"); you may not use this documentation except in
+ compliance with the Documentation License. You may obtain a copy of the
+ Documentation License at
+
+ https://creativecommons.org/licenses/by/4.0/
+
+ Unless required by applicable law or agreed to in writing, documentation
+ distributed under the Documentation License is distributed on an "AS IS"
+ BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied. See the Documentation License for the specific language governing
+ permissions and limitations under the Documentation License.
--- /dev/null
+# vim: ts=4 sw=4 noet:
+
+# -------------------------------------------------------------------------------
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# -------------------------------------------------------------------------------
+
+# CI to verify the MC application components
+# Inherits C toolchain from buildpack-deps:stretch then adds cmake and better shell(s).
+
+# It is assumed that this docker file is used with a build command run at the
+# root level of the repo (directory containing the ci directory). E.g.
+# docker build -f ci/Dockerfile .
+
+FROM buildpack-deps:stretch
+
+RUN apt-get update && apt-get -q -y install cmake ksh
+
+# stuff our repo things into a scratch area
+RUN mkdir /playpen
+ADD . /playpen
+
+
+# add any unit test scripts that need to be driven, e.g.
+# RUN ksh test/mcl_unit_test.ksh
+
+# This is a final/only script which might print useful things to the log and ALWAYS succeeds.
+RUN ksh /playpen/ci/stats.ksh
+
+# there is no cmd; the build/verification assumes that if the image is created
+# successfully, e.g. none of the previous run commands fail, that the environment
+# is successfully vetted.
--- /dev/null
+ -------------------------------------------------------------------------------
+ Copyright (c) 2018-2019 AT&T Intellectual Property.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ -------------------------------------------------------------------------------
+
+These files are needed to support the continuous integration process
+which verifies code on commit and might build images for the application.
+
+The Docker file is used to build an image during CI vetting, and maybe even
+during the building of the overall application container. For vetting,
+the successful build indicates that all vetting of the repo passed.
--- /dev/null
+
+# vim: ts=4 sw=4 noet:
+
+# -------------------------------------------------------------------------------
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# -------------------------------------------------------------------------------
+
+# -------------------------------------------------------------------------------
+# Mnemonic: stats.ksh
+# Abstract: This script is run as the last vetting script during the CI
+# process to possibly provide some stats about the verification.
+# The script may be run as the only verification script during
+# initial development. The script MUST always finish successfully
+# Date: 24 August 2019
+# -------------------------------------------------------------------------------
+
+
+echo "$(date) finished verification and/or build"
--- /dev/null
+
+# this provides the jenkins environment with a tag to use when
+# building verification containers.
+---
+tag: 1.0.0
--- /dev/null
+# vim: ts=4 sw=4 noet:
+#----------------------------------------------------------------------------------
+#
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#---------------------------------------------------------------------------------
+
+# this make file assuems that both NNG and RMR are installed and that the variables
+# LD_LIBRARY_PATH, LIBRARY_PATH are set correctly.
+
+binaries = mc_listener
+test_progs = sender unit_test pipe_reader
+lib_obj = mcl.o
+lib_h = mcl.h
+
+coverage_opts = -ftest-coverage -fprofile-arcs
+
+# make with no parms should just build 'production' binaries
+all: $(binaries) mc_listener
+
+libmcl.a: $(lib_obj) $(lib_h)
+ ar -v -r libmcl.a $<
+
+mc_listener: mc_listener.c libmcl.a
+ gcc mc_listener.c -o mc_listener -L. -lmcl -lrmr_nng -lnng -lm -lpthread
+
+
+# ---- testing stuff -----------------------------------------------------------------
+tests: $(test_progs)
+
+sender : sender.c
+ gcc sender.c -o sender -lrmr_nng -lnng -lm -lpthread
+
+pipe_reader : pipe_reader.c libmcl.a
+ gcc pipe_reader.c -o pipe_reader -L. -lmcl -lrmr_nng -lnng -lm -lpthread
+
+unit_test: unit_test.c mcl.c
+ gcc $(coverage_opts) unit_test.c -o unit_test -lrmr_nng -lnng -lm -lpthread
+
+
+# ---- housekeeping stuff -------------------------------------------------------------
+# remove only intermediates
+clean:
+ rm -f *.o *.gcda *.gcno *.gcov
+
+# remove anything that can be rebuilt
+nuke: clean
+ rm -f *mcl.a $(binaries) $(test_progs)
--- /dev/null
+
+--------------------------------------------------------------------------------
+
+ Copyright (c) 2018-2019 AT&T Intellectual Property.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+--------------------------------------------------------------------------------
+
+
+
+MC Listener
+
+This directory contains the source for the a simple message listener which
+writes messages received via RMR into a fifo (named pipe) for an external
+process to consume.
+
+Fifos are named MT_xxxxxxxx where the Xs are replaced with the message
+type with up to 7 leading zeros (e.g. MT_00000002). The data written
+to a pipe has the form:
+ <8 bytes><n bytes>
+
+Where the first 8 bytes are the ASCII representation of the length of
+the message (n) (the 8th byte is a zero allowing the bytes to be
+treated as a C string if desired. The next n bytes are the unchanged
+payload which was received. No RMR header information (e.g. source,
+meid, etc.) is communicated.
+
+If the listener is executed with the timestamp extension enabled, then
+the leading 'header' is enhanced such that the time the message was
+received by the listener is added. Additionally, a leading delimiter
+is added to make synchronisation possible. The timestamp is also an
+ASCII string of the form 1570110224356 (milliseconds past the epoch).
+The enhanced header has the format:
+
+ <delimeter><length-bytes><time-bytes>
+
+The <delimeter> is a series of 4 bytes which should always be: '@MCL'.
+It is intended to be used to sequence "frames" in the pipe should there be
+write errors which result in missing data. If the application reading from
+a pipe does not see this delimeter, then it should read byte by byte from
+the pipe until it does in order to synchronise with the stream.
+
+The <length-bytes> are as descrbed previously: 8 byte ASCII string (nil
+terminated).
+
+The <time-bytes> is an ASCII string (nil terminated) with a length of 16
+bytes.
+
+The entire header will require 28 bytes.
+
+
+There are multiple docker files; *.df.
+ mcl_runime.df -- builds an image with the runtime mc_listener binary
+ mcl_dev.df -- builds a development image that can be used to
+ interactively build and test the library and mc_listener
+ application.
+
+Unit testing
+A very small set of unit tests are provided for the library functions in
+mcl.c. Because of the nature of the fanout function, which blocks waiting
+on RMR messages, it is not possible to unit test that bit of code.
+
+FIFO Reader
+The pipe_reader programme is a simple application which uses the mcl.c
+library functions to open and read from a single pipe. If the -e option
+is given it will expect that data in the FIFO has extended headers. Use
+the -? option (or -h) to generate a full usage statement.
--- /dev/null
+#!/usr/bin/env bash
+# vim: ts=4 sw=4 noet:
+#--------------------------------------------------------------------------------
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#--------------------------------------------------------------------------------
+
+
+
+# ---------------------------------------------------------------------------
+# Mnemonic: bld_dev_env.sh
+# Abstract: This script is meant to be executed by a RUN command in
+# a docker file. It fetches NNG, builds/installs it, and then
+# does a wget of the desired RMR version's packages and installs
+# them. RMR is fetched from package cloud.
+#
+# Date: 22 August 2019
+# Author: E. Scott Daniels
+# ---------------------------------------------------------------------------
+
+rmr_ver=1.9.0
+nng_ver=v1.1.1
+
+while [[ $1 == -* ]]
+do
+ case $1 in
+ -n) nng_ver=$2; shift;;
+ -r) rmr_ver=$2; shift;;
+ esac
+
+ shift
+done
+
+if [[ $nng_ver != "v"* ]]
+then
+ nng_ver="v$nng_ver"
+fi
+
+set -e # from this point on crash on any error
+mkdir -p /playpen/build/nng
+cd /playpen/build/nng
+git clone https://github.com/nanomsg/nng.git
+cd nng
+git checkout $nng_ver
+mkdir .build
+cd .build
+echo "building nng (messages supressed unless there is an error)"
+
+set +e
+if ! cmake .. >/tmp/cmake.nng.log 2>&1
+then
+ cat /tmp/cmake.nng.log
+ echo ""
+ echo "### ERROR ### NNG cmake configuration failed"
+ exit 1
+fi
+
+if ! make install >/tmp/build.nng.log 2>&1
+then
+ cat /tmp/build.nng.log
+ echo ""
+ echo "### ERROR ### NNG build failed"
+ exit 1
+fi
+
+set -e
+echo "nng build finished ok"
+cd /playpen
+rm -fr /playpen/build/nng
+
+echo "installing RMR packages version = $rmr_ver"
+mkdir -p /playpen/build/pkgs
+cd /playpen/build/pkgs
+
+base_url=https://packagecloud.io/o-ran-sc/master/packages/debian/stretch/
+base_url=https://packagecloud.io/o-ran-sc/staging/packages/debian/stretch/
+pc_url=${base_url}rmr_${rmr_ver}_amd64.deb/download.deb
+pc_dev_url=${base_url}rmr-dev_${rmr_ver}_amd64.deb/download.deb
+
+wget -q -O rmr.deb $pc_url
+wget -q -O rmr-dev.deb $pc_dev_url
+
+ls -al
+dpkg -i *.deb
+cd /playpen
+rm -fr /playpen/build/pkgs
+echo "RMR package install finished"
--- /dev/null
+#!/usr/bin/env bash
+# vim: ts=4 sw=4 noet:
+
+#--------------------------------------------------------------------------------
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#--------------------------------------------------------------------------------
+
+#------------------------------------------------------------------------------------
+# Mnemonic: build_images.sh
+# Abstract: This script will create both the mc_listener runtime and development
+# images.
+# Date: 22 August 2-19
+# Author: E. Scott Daniels
+# -----------------------------------------------------------------------------------
+
+skip_dev=1
+if [[ $1 == "all" ]]
+then
+ skip_dev=0
+ shift
+fi
+
+if [[ $1 == "-?" || $1 == "-h" ]]
+then
+ echo "usage: $0 [all] [mcl-version-tag [patch-level]]"
+ echo " using all as first keyword causes both runtime and dev images to build"
+ exit 0
+fi
+
+
+ver=${1:-1.1}
+patch=${2:-0}
+
+if (( skip_dev == 0 ))
+then
+ echo "building development image"
+ docker build -f mcl_dev.df -t mcl_dev:$ver.$patch .
+fi
+
+echo "building runtime image mc_listener:$ver"
+if docker build -f mcl_runtime.df -t mc_listener:$ver.$patch .
+then
+ echo "build finished"
+ echo ""
+ docker images|grep mc_
+fi
+
--- /dev/null
+// vim: ts=4 sw=4 noet:
+/*
+--------------------------------------------------------------------------------
+ Copyright (c) 2018-2019 AT&T Intellectual Property.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+--------------------------------------------------------------------------------
+*/
+
+/*
+ Mnemonic: mc_listener.c
+ Abstract: This application (management campaign listener) will listen for
+ RMR based messages and write the payloads into FIFOs which
+ correspond to the message type.
+
+ Defaults:
+ /var/lib/mc/listener -- directory for FIFOs
+
+ Command line options:
+ -d <path> FIFO directory (default is /tmp/mcl/fifos)
+ -p <port> The port to set RMR listener on (default is 4560)
+ -r <seconds> The frequency that count reports are written to
+ stderr. 0 == 0ff; default is 60.
+
+
+ RMR based environment variables which might be needed:
+ RMR_SEED_RT -- path to the static routing table
+ RMR_RTG_SVC -- port to listen for RTG connections
+
+ Date: 22 August 2019
+ Author: E. Scott Daniels
+*/
+
+#include <unistd.h>
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <time.h>
+#include <string.h>
+
+
+#include "mcl.h"
+
+//---- support -----------------------------------------------------------------------------
+
+static void bad_arg( char* what ) {
+ fprintf( stderr, "[ERR] option is unrecognised or isn't followed by meaningful data: %s\n", what );
+}
+
+static void usage( char* argv0 ) {
+ fprintf( stderr, "usage: %s [-d fifo-dir] [-e] [-p listen-port] [-q | -r report-freq]\n", argv0 );
+ fprintf( stderr, " -e enable extended header in buffers written to FIFOs\n" );
+}
+
+//------------------------------------------------------------------------------------------
+int main( int argc, char** argv ) {
+ void* ctx; // the mc listener library context
+ char* dname = "/tmp/mcl/fifos"; // default directory where we open fifos
+ char* port = "4560"; // default rmr port
+ char* siphon_dir = "/tmp/mci/siphon"; // where siphon files are placed
+ int siphon = 0; // percentage of messages to siphone off
+ int report_freq = 60; // report stats every n seconds
+ int pidx = 1; // parameter index
+ int error = 0;
+ int long_hdrs = 0; // -e sets and causes extended headers to be written
+
+ while( pidx < argc && argv[pidx][0] == '-' ) { // simple argument parsing (-x or -x value)
+ switch( argv[pidx][1] ) {
+ case 'd':
+ if( pidx+1 < argc ) {
+ dname = strdup( argv[pidx+1] );
+ pidx++;
+ } else {
+ bad_arg( argv[pidx] );
+ error = 1;
+ }
+ break;
+
+ case 'e':
+ long_hdrs = 1;
+ break;
+
+ case 'p':
+ if( pidx+1 < argc ) {
+ port = strdup( argv[pidx+1] );
+ pidx++;
+ } else {
+ bad_arg( argv[pidx] );
+ error = 1;
+ }
+ break;
+
+ case 'q':
+ report_freq = 0;
+ break;
+
+ case 'r':
+ if( pidx+1 < argc ) {
+ report_freq = atoi( argv[pidx+1] );
+ pidx++;
+ } else {
+ bad_arg( argv[pidx] );
+ error = 1;
+ }
+ break;
+
+ case 'h':
+ case '?':
+ usage( argv[0] );
+ exit( 0 );
+ break;
+
+ default:
+ bad_arg( argv[pidx] );
+ error = 1;
+ break;
+ }
+
+ pidx++;
+ }
+
+ if( error ) {
+ usage( argv[0] );
+ exit( 1 );
+ }
+
+ ctx = mcl_mk_context( dname ); // initialise the library context
+ if( ctx == NULL ) {
+ fprintf( stderr, "[FAIL] couldn't initialise the mc listener environment\n" );
+ exit( 1 );
+ }
+ mcl_set_sigh(); // set signal handler(s)
+
+ mcl_start_listening( ctx, port, MCL_NOWAIT ); // start the listener, no waiting for rt since we don't send
+ mcl_fifo_fanout( ctx, report_freq, long_hdrs ); // listen and fanout messages to fifo; report to stdout every ~2sec
+
+ fprintf( stderr, "[INFO] mc listener is finished.\n" );
+}
+
--- /dev/null
+// vim: ts=4 sw=4 noet:
+/*
+--------------------------------------------------------------------------------
+ Copyright (c) 2018-2019 AT&T Intellectual Property.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+--------------------------------------------------------------------------------
+*/
+
+/*
+ Mnemonic: mcl.c.
+ Abstract: The mc listener library content. All external functions
+ should start with mcl_ and all stderr messages should have
+ (mcl) as the first token following the severity indicator.
+
+ Date: 22 August 2019
+ Author: E. Scott Daniels
+*/
+
+#include <unistd.h>
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <time.h>
+#include <string.h>
+#include <fcntl.h>
+#include <signal.h>
+#include <sys/stat.h>
+
+#include <rmr/rmr.h>
+#include <rmr/rmr_symtab.h>
+
+#include "mcl.h"
+
+#define READER 0
+#define WRITER 1
+
+#define TRUE 1
+#define FALSE 0
+
+
+/*
+ Information about one file descriptor. This is pointed to by the hash
+ such that the message type can be used as a key to look up the fifo's
+ file descriptor.
+*/
+typedef struct {
+ int fd; // open fdes
+ int key; // symtab key
+ long long wcount; // number of writes
+ long long drops; // number dropped
+
+ long long wcount_rp; // number of writes during last reporting period
+ long long drops_rp; // number dropped during last reporting period
+} fifo_t;
+
+/*
+ Our conext. Pointers to the read and write hash tables (both keyed on the message
+ type), the message router (RMR) context, and other goodies.
+*/
+typedef struct {
+ void* mrc; // the message router's context
+ void* wr_hash; // symtable to look up pipe info based on mt for writing
+ void* rd_hash; // we support reading from pipes, but need a different FD for that
+ char* fifo_dir; // directory where we open fifos
+
+} mcl_ctx_t;
+
+// -------- private -------------------------------------------------------
+
+
+/*
+ Builds an extended header in the buffer provided, or allocates a new buffer if
+ dest is nil. The header is of the form:
+ <delim><len><timestamp>
+
+ Timestamp is a single unsigned long long in ASCII; ms since epoch.
+ If the current time is 2019/10/03 10:39:51.103 which is 1570113591.103
+ the timestamp generated will be 1570113591103.
+*/
+static char* build_hdr( int len, char* dest, int dest_len ) {
+ struct timespec ts; // time just before call executed
+
+ if( dest == NULL ) {
+ dest_len = 48;
+ dest = (char *) malloc( sizeof( char ) * dest_len );
+ } else {
+ if( dest_len < 28 ) { // shouldn't happen, but take no chances
+ memset( dest, 0, dest_len );
+ return NULL;
+ }
+ }
+
+ memset( dest, 0, dest_len );
+
+ clock_gettime( CLOCK_REALTIME, &ts );
+ sprintf( dest, "%s%07d", MCL_DELIM, len );
+ sprintf( dest+12, "%ld%03ld", ts.tv_sec, ts.tv_nsec/1000000 );
+
+ return dest;
+}
+
+/*
+ Build a file name and open. The io_direction is either READER or
+ WRITER. For a writer we must 'trick' the system into allowing us
+ to open a pipe for writing in non-blocking mode so that we can
+ report on drops (messages we couldn't write because there was no
+ reader). The trick is to open a reader on the pipe so that when
+ we open the writer there's a reader and the open won't fail. As
+ soon as we have the writer open, we can close the junk reader.
+
+ If the desired fifo does not exist, it is created.
+*/
+static int open_fifo( mcl_ctx_t* ctx, int mtype, int io_dir ) {
+ char wbuf[1024];
+ int fd; // real file des
+ int jfd = -1; // junk file des
+ int state;
+
+ if( ctx == NULL || mtype < 0 ) {
+ return -1;
+ }
+
+ snprintf( wbuf, sizeof( wbuf ), "%s/MT_%09d", ctx->fifo_dir, mtype );
+
+ state = mkfifo( wbuf, 0660 ); // make the fifo; this will fail if it exists and that's ok
+ if( state != 0 && errno != EEXIST ) {
+ fprintf( stderr, "[ERR] (mcl) unable to create fifo: %s: %s\n", wbuf, strerror( errno ) );
+ return -1;
+ }
+
+ if( io_dir == READER ) {
+ fd = open( wbuf, O_RDONLY ); // just open the reader
+ if( fd < 0 ) {
+ fprintf( stderr, "[ERR] (mcl) fifo open failed (ro): %s: %s\n", wbuf, strerror( errno ) );
+ }
+ } else {
+ jfd = open( wbuf, O_RDWR | O_NONBLOCK ); // must have a reader before we can open a non-blocking writer
+ if( jfd < 0 ) {
+ fprintf( stderr, "[ERR] (mcl) fifo open failed (rw): %s: %s\n", wbuf, strerror( errno ) );
+ }
+
+ fd = open( wbuf, O_WRONLY | O_NONBLOCK ); // this will be our actual writer, in non-blocking mode
+ if( fd < 0 ) {
+ fprintf( stderr, "[ERR] (mcl) fifo open failed (wo): %s: %s\n", wbuf, strerror( errno ) );
+ }
+
+ close( jfd ); // should be safe to close this
+ }
+
+
+ return fd;
+}
+
+/*
+ Given a message type, return the file des of the fifo that
+ the payload should be written to. Returns the file des, or -1
+ on error. When sussing out a read file descriptor this will
+ block until there is a fifo for the message type which is
+ open for reading.
+
+ If fref is not nil, then a pointer to the fifo info block is returned
+ allowing for direct update of counts after the write.
+*/
+static int suss_fifo( mcl_ctx_t* ctx, int mtype, int io_dir, fifo_t** fref ) {
+ fifo_t* fifo;
+ void* hash;
+
+ if( io_dir == READER ) { // with an integer key, we nned two hash tables
+ hash = ctx->rd_hash;
+ } else {
+ hash = ctx->wr_hash;
+ }
+
+ if( (fifo = (fifo_t *) rmr_sym_pull( hash, mtype )) == NULL ) {
+ fifo = (fifo_t *) malloc( sizeof( *fifo ) );
+ if( fifo == NULL ) {
+ return -1;
+ }
+
+ memset( fifo, 0, sizeof( *fifo ) );
+ fifo->key = mtype;
+ fifo->fd = open_fifo( ctx, mtype, io_dir );
+ rmr_sym_map( hash, mtype, fifo );
+ }
+
+ if( fref != NULL ) {
+ *fref = fifo;
+ }
+ return fifo->fd;
+}
+
+/*
+ Make marking counts easier in code
+*/
+static inline void chalk_error( fifo_t* fifo ) {
+ if( fifo != NULL ) {
+ fifo->drops++;
+ fifo->drops_rp++;
+ }
+}
+
+static inline void chalk_ok( fifo_t* fifo ) {
+ if( fifo != NULL ) {
+ fifo->wcount++;
+ fifo->wcount_rp++;
+ }
+}
+
+/*
+ Callback function driven to traverse the symtab and generate the
+ counts for each fifo.
+*/
+static void wr_stats( void* st, void* entry, char const* name, void* thing, void* data ) {
+ fifo_t* fifo;
+ int report_period = 60;
+
+ if( data ) {
+ report_period = *((int *) data);
+ }
+
+ if( (fifo = (fifo_t *) thing) != NULL ) {
+ fprintf( stdout, "[STAT] (mcl) mtype=%d total writes=%lld total drops=%lld; during last %ds writes=%lld drops=%lld\n",
+ fifo->key, fifo->wcount, fifo->drops, report_period, fifo->wcount_rp, fifo->drops_rp );
+
+ fifo->wcount_rp = 0; // reset the report counts
+ fifo->drops_rp = 0;
+ }
+}
+
+// ---------- public ------------------------------------------------------
+/*
+ Sets a signal handler for sigpipe so we don't crash if a reader closes the
+ last reading fd on a pipe. We could do this automatically, but if the user
+ programme needs to trap sigpipe too, this gives them the option not to have
+ us interfere.
+*/
+extern int mcl_set_sigh( ) {
+ signal( SIGPIPE, SIG_IGN );
+}
+
+/*
+ "Opens" the interface to RMR such that messages sent to the application will
+ be available via the rmr receive funcitons. This is NOT automatically called
+ by the mk_context function as some applications will be using the mc library
+ for non-RMR, fifo, chores.
+*/
+extern int mcl_start_listening( void* vctx, char* port, int wait4ready ) {
+ mcl_ctx_t* ctx;
+ int announce = 0;
+
+ if( (ctx = (mcl_ctx_t*) vctx) == NULL ) {
+ return 0;
+ }
+
+ ctx->mrc = rmr_init( port, RMR_MAX_RCV_BYTES, RMRFL_NONE ); // start your engines!
+ if( ctx->mrc == NULL ) {
+ fprintf( stderr, "[CRIT] unable to initialise RMr\n" );
+ return 0;
+ }
+
+ while( wait4ready && ! rmr_ready( ctx->mrc ) ) { // only senders need to wait
+ if( announce <= 0 ) {
+ fprintf( stderr, "[INFO] waiting for RMR to show ready\n" );
+ announce = 10;
+ } else {
+ announce--;
+ }
+
+ sleep( 1 );
+ }
+
+ return 1;
+}
+
+/*
+ Blocks until a message arives with a good return code or we timeout. Returns the
+ rmr message buffer. Timeout value epxected in seconds.
+*/
+extern rmr_mbuf_t* mcl_get_msg( void* vctx, rmr_mbuf_t* msg, int timeout ) {
+ mcl_ctx_t* ctx;
+
+ if( (ctx = (mcl_ctx_t *) vctx) == NULL ) {
+ return NULL;
+ }
+
+ if( ctx->mrc == NULL ) {
+ fprintf( stderr, "bad context\n" );
+ exit( 1 );
+ }
+
+ do {
+ msg = rmr_torcv_msg( ctx->mrc, msg, timeout * 1000 ); // wait for next
+ } while( msg == NULL || (msg->state != RMR_OK && msg->state != RMR_ERR_TIMEOUT) );
+
+ return msg;
+}
+
+/*
+ Create the context.
+*/
+extern void* mcl_mk_context( char* dir ) {
+ mcl_ctx_t* ctx;
+
+ if( (ctx = (mcl_ctx_t *) malloc( sizeof( *ctx ) )) != NULL ) {
+ memset( ctx, 0, sizeof( *ctx ) );
+ ctx->fifo_dir = strdup( dir );
+ ctx->wr_hash = rmr_sym_alloc( 1001 );
+ ctx->rd_hash = rmr_sym_alloc( 1001 );
+
+ if( ctx->wr_hash == NULL || ctx->rd_hash == NULL ) {
+ fprintf( stderr, "[ERR] (mcl) unable to allocate hash table for fifo keys\n" );
+ free( ctx );
+ return NULL;
+ }
+ }
+
+ return (void *) ctx;
+}
+
+/*
+ Read the header. Best case we read the expected number of bytes, get all
+ of them and find that they start with the delemiter. Worst case
+ We have to wait for all of the header, or need to synch at the next
+ delimeter. We assume best case most likely and handle it as such.
+*/
+static void read_header( int fd, char* buf ) {
+ int len;
+ int need = MCL_EXHDR_SIZE; // total needed
+ int dneed; // delimieter needed
+ int rlen;
+ char* rp; // read position in buf
+
+ len = read( fd, buf, need );
+ if( len == need && strncmp( buf, MCL_DELIM, strlen( MCL_DELIM )) == 0 ) { // best case, most likely
+ return;
+ }
+
+ while( TRUE ) {
+ if( len < strlen( MCL_DELIM ) ) { // must get at least enough bytes to check delim
+ rp = buf + len;
+ dneed = strlen( MCL_DELIM ) - len;
+
+ while( dneed > 0 ) {
+ len = read( fd, rp, dneed );
+ dneed -= len;
+ rp += len;
+ }
+ }
+
+ if( strncmp( buf, MCL_DELIM, strlen( MCL_DELIM )) == 0 ) { // have a good delimiter, just need to wait for bytes
+ need = MCL_EXHDR_SIZE - strlen( MCL_DELIM );
+ rp = buf + (MCL_EXHDR_SIZE - need);
+
+ while( need > 0 ) {
+ len = read( fd, rp, need );
+ need -= len;
+ rp += len;
+ }
+
+ return;
+ }
+
+ while( buf[0] != MCL_DELIM[0] ) { // wait for a recognised start byte to be read (may cause an additional message drop
+ len = read( fd, buf, 1 ); // because we ignore start byte that might be in the buffer)
+ }
+
+ need = MCL_EXHDR_SIZE - len;
+ }
+}
+
+
+/*
+ Read one record from the fifo that the message type maps to.
+ Writes at max ublen bytes into the ubuf.
+
+ If long_hdrs is true (!0), then we expect that the stream in the fifo
+ has extended headers (<delim><len><time>), and will write the timestamp
+ from the header into the buffer pointed to by timestamp. The buffer is
+ assumed to be at least MCL_TSTAMP_SIZE bytes in length.
+
+ Further, when extended headers are being used, this function will
+ automatically resynchronise if it detects an issue.
+
+ The function could look for the delimiter and automatically detect whether
+ or not extended headers are in use, but if the stream is out of synch on the
+ first read, this cannot be done, so the funciton requires that the caller
+ know that the FIFO contains extended headers.
+*/
+static int fifo_read1( void *vctx, int mtype, char* ubuf, int ublen, int long_hdrs, char* timestamp ) {
+ int fd;
+ int len;
+ int msg_len;
+ int got = 0; // number of bytes we actually got
+ int need;
+ char wbuf[4096];
+ mcl_ctx_t* ctx; // our context; mostly for the rmr context reference and symtable
+ fifo_t* fref = NULL; // the fifo struct we found
+
+ if( (ctx = (mcl_ctx_t*) vctx) == NULL ) {
+ errno = EINVAL;
+ return 0;
+ }
+
+ if( (fd = suss_fifo( ctx, mtype, READER, NULL )) >= 0 ) {
+ if( long_hdrs ) {
+ read_header( fd, wbuf );
+ msg_len = need = atoi( wbuf + MCL_LEN_OFF ); // read the length
+ if( timestamp ) {
+ strcpy( timestamp, wbuf + MCL_TSTAMP_OFF+1 );
+ }
+ } else {
+ if( timestamp != NULL ) { // won't be there, but ensure it's not garbage
+ *timestamp = 0;
+ }
+
+ len = read( fd, wbuf, MCL_LEN_SIZE ); // we assume we will get all 8 as there isn't a way to sync the old stream
+ msg_len = need = atoi( wbuf );
+ }
+
+
+ if( need > ublen ) {
+ need = ublen; // cannot give them more than they can take
+ }
+ while( need > 0 ) {
+ len = read( fd, wbuf, need > sizeof( wbuf ) ? sizeof( wbuf ) : need );
+ memcpy( ubuf+got, wbuf, len );
+ got += len;
+ need -= len;
+ }
+
+ if( msg_len > got ) { // we must ditch rest of this message
+ need = msg_len = got;
+ while( need > 0 ) {
+ len = read( fd, wbuf, need > sizeof( wbuf ) ? sizeof( wbuf ) : need );
+ need -= len;
+ }
+ }
+
+ return got;
+ }
+
+ errno = EBADFD;
+ return 0;
+}
+
+/*
+ Read one record from the fifo that the message type maps to.
+ Writes at max ublen bytes into the ubuf. If extended headers are in use
+ this function will ignore the timestamp.
+
+ If long_hdrs is true (!0), then we expect that the stream in the fifo
+ has extended headers (<delim><len><time>).
+*/
+extern int mcl_fifo_read1( void *vctx, int mtype, char* ubuf, int ublen, int long_hdrs ) {
+ return fifo_read1( vctx, mtype, ubuf, ublen, long_hdrs, NULL );
+}
+
+/*
+ Read a single message from the FIFO returning it in the caller's buffer. If extended
+ headers are being used, and the caller supplied a timestamp buffer, the timestamp
+ which was in the header will be returned in that buffer. The return value is the number
+ of bytes in the buffer; 0 indicates an error and errno should be set.
+*/
+extern int mcl_fifo_tsread1( void *vctx, int mtype, char* ubuf, int ublen, int long_hdrs, char *timestamp ) {
+ return fifo_read1( vctx, mtype, ubuf, ublen, long_hdrs, timestamp );
+}
+
+
+/*
+ Will read messages and fan them out based on the message type. This should not
+ return and if it does the caller should assume an error.
+
+ The output to each fifo is MCL_LEN_SIZE bytes with an ASCII, zero terminated, length
+ string , followed by that number of 'raw' bytes. The raw bytes are the payload
+ exactly as received.
+
+ The report parameter is the frequency, in seconds, for writing a short
+ status report to stdout. If 0 then it's off.
+
+ If long_hdr is true, then we geneate an extended header with a delimiter and
+ timestamp.
+*/
+extern void mcl_fifo_fanout( void* vctx, int report, int long_hdr ) {
+ mcl_ctx_t* ctx; // our context; mostly for the rmr context reference and symtable
+ fifo_t* fifo; // fifo to chalk counts on
+ rmr_mbuf_t* mbuf = NULL; // received message buffer; recycled on each call
+ char wbuf[128]; // buffer to build len string in
+ int state;
+ int fd; // file des to write to
+ long long total = 0; // total messages received and written
+ long long total_drops = 0; // total messages received and written
+ long count = 0; // messages received and written during last reporting period
+ long errors = 0; // unsuccessful payload writes
+ long drops; // number of drops
+ time_t next_report = 0; // we'll report every 2 seconds if report is true
+ time_t now;
+ int hwlen; // write len for header
+
+ if( (ctx = (mcl_ctx_t*) vctx) == NULL ) {
+ fprintf( stderr, "[ERR] (mcl) invalid context given to fanout\n" );
+ errno = EINVAL;
+ return;
+ }
+
+ if( report < 0 ) {
+ report = 0;
+ }
+
+ while( 1 ) {
+ mbuf = mcl_get_msg( ctx, mbuf, report ); // wait up to report sec for msg (0 == block until message)
+
+ if( mbuf != NULL && mbuf->state == RMR_OK && mbuf->len > 0 ) {
+ fd = suss_fifo( ctx, mbuf->mtype, WRITER, &fifo ); // map the message type to an open fd
+ if( fd >= 0 ) {
+ if( long_hdr ) {
+ build_hdr( mbuf->len, wbuf, sizeof( wbuf ) );
+ hwlen = MCL_EXHDR_SIZE;
+ } else {
+ snprintf( wbuf, sizeof( wbuf ), "%07d", mbuf->len ); // size of payload CAUTION: 7d is MCL_LEN_SIZE-1
+ hwlen = MCL_LEN_SIZE;
+ }
+
+ if( (state = write( fd, wbuf, hwlen )) != hwlen ) { // write exactly MCL_LEN_SIZE bytes from the buffer
+ drops++;
+ total_drops++;
+ chalk_error( fifo );
+ } else {
+ if( write( fd, mbuf->payload, mbuf->len ) < mbuf->len ) { // followed by the payload
+ errors++;
+ chalk_error( fifo );
+ } else {
+ chalk_ok( fifo );
+ count++;
+ total++;
+ }
+ }
+ }
+ }
+
+ if( report ) {
+ if( (now = time( NULL ) ) > next_report ) {
+ rmr_sym_foreach_class( ctx->wr_hash, 0, wr_stats, &report ); // run endpoints in the active table
+ fflush( stdout );
+
+ fprintf( stdout, "[STAT] (mcl) total writes=%lld total drops=%lld; during last %ds writes=%ld drops=%ld errs=%ld errors\n",
+ total, total_drops, report, count, drops, errors );
+ next_report = now + report;
+ count = 0;
+ drops = 0;
+
+ fflush( stdout );
+ }
+ }
+ }
+}
+
+
--- /dev/null
+// vim: ts=4 sw=4 noet:
+/*
+ --------------------------------------------------------------------------------
+ Copyright (c) 2018-2019 AT&T Intellectual Property.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ --------------------------------------------------------------------------------
+
+/*
+ Mnemonic: mcl.h
+ Abstract: Header file for the mc listener library.
+ Date: 22 August 2019
+ Author: E. Scott Daniels
+
+*/
+
+#ifndef mcl_h
+#define mcl_h
+
+#include <rmr/rmr.h>
+#include <rmr/rmr_symtab.h>
+
+// ------- public constants and structs -------------------------------------------------
+
+#define MCL_LEN_SIZE 8 // number of bytes the length has in both short and extended header
+#define MCL_DELIM_SIZE 4 // number of bytes in extended header delimiter
+#define MCL_TSTAMP_SIZE 16 // number of bytes in extended header timestamp
+#define MCL_EXHDR_SIZE (MCL_DELIM_SIZE+MCL_LEN_SIZE+MCL_TSTAMP_SIZE) // total size of extended header
+
+#define MCL_DELIM "@MCL" // delimeter to synch msgs in fifo
+
+#define MCL_TSTAMP_OFF (MCL_DELIM_SIZE+MCL_LEN_SIZE) // offsets in header
+#define MCL_LEN_OFF MCL_DELIM_SIZE
+
+#define MCL_NOWAIT 0 // do not wait for RMR route table to arrive
+#define MCL_WAIT 1 // block reader start until RMR route table is initialised
+
+//------------ prototypes --------------------------------------------------------------
+extern void mcl_fifo_fanout( void* ctx, int report, int long_hdrs );
+extern rmr_mbuf_t* mcl_get_msg( void* vctx, rmr_mbuf_t* msg, int timeout );
+extern void* mcl_mk_context( char* dir );
+extern int mcl_fifo_read1( void* vctx, int mtype, char* ubuf, int ublen, int long_hdr );
+extern int mcl_fifo_tsread1( void* vctx, int mtype, char* ubuf, int ublen, int long_hdr, char* timestamp );
+extern int mcl_set_sigh( );
+extern int mcl_start_listening( void* vctx, char* port, int wait4ready );
+
+#endif
--- /dev/null
+# vim: ts=4 sw=4 noet:
+#==================================================================================
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#==================================================================================
+
+
+# Mnemonic: mcl_dev.df
+# Abstract: This is a docker build file which creates an image that can be used
+# for interactive development of the MC listener and related applications.
+# Building should be as simple as:
+# docker build -f mcl_dev.df -t mcl_dev:latest .
+#
+# The image contains the NNG and RMR libraries and headers needed. The
+# MC listener source is NOT placed into the image. Rather the source should
+# be mounted as a volume when the container is created:
+#
+# docker run --rm -it -v $PWD:/playpen/src -u <uid>:<gid> sh
+# (assuming that the source is in the current working directory).
+# Once the container is running, vi, make and git are all available.
+# Date: 22 August 2019
+# Author: E. Scott Daniels
+#---------------------------------------------------------------------------------------
+
+FROM ubuntu:18.04
+RUN mkdir /playpen
+
+RUN apt-get update && apt-get install -y cmake gcc make vim git
+RUN apt-get install -y cmake g++ wget
+
+RUN mkdir /playpen/bin
+COPY build_dev_env.sh /playpen/bin/
+RUN bash /playpen/bin/build_dev_env.sh
+
+ENV LD_LIBRARY_PATH=/usr/local/lib;/usr/local/lib64
+ENV C_INCLUDE_PATH=/usr/local/include
+
+CMD [ "bash" ]
+
+
--- /dev/null
+# vim: ts=4 sw=4 noet:
+#==================================================================================
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#==================================================================================
+
+
+# Mnemonic: mcl_runtime.df
+# Abstract: Docker build file which creaes a runtime image for the listener.
+# Building should be as simple as:
+# docker build -f mcl_runtime.f -t mc_listener:xx.xx .
+#
+# Running the image in a container:
+# The fifo output directory needs to be mounted to the container
+# e.g. -v /exports/mcl/fifos:/tmp/mcl/fifos. The internal
+# directory can be supplied on the command line using -d path, but
+# should not be needed.
+#
+# The RMR listen port must be properly exposed to the container. The
+# internal default is 4560, and can be changed with the -p port option
+# on the command line (needed if running from docker-compose). Typical
+# exposure might be -p <host-port>:4560 on the docker run command.
+#
+# The MC listener application does NOT need route table updates and thus
+# there is NO need to expose the route table generator port on this
+# container.
+#
+# Date: 22 August 2019
+# Author: E. Scott Daniels
+
+
+FROM ubuntu:18.04 as buildenv
+RUN mkdir /playpen
+
+RUN apt-get update && apt-get install -y cmake gcc make git g++ wget
+
+RUN mkdir /playpen/bin /playpen/src
+COPY build_dev_env.sh /playpen/bin/
+COPY Makefile *.h *.c /playpen/src/
+
+ENV LD_LIBRARY_PATH=/usr/local/lib64:/usr/local/lib
+ENV C_INCLUDE_PATH=/usr/local/include
+RUN bash /playpen/bin/build_dev_env.sh
+RUN cd /playpen/src/; make -B mc_listener sender pipe_reader
+
+# ----- final, smaller image ----------------------------------
+FROM ubuntu:18.04
+
+RUN mkdir -p /playpen/bin
+COPY --from=buildenv /usr/local/lib/* /usr/local/lib/
+COPY --from=buildenv /playpen/src/mc_listener /playpen/src/sender /playpen/src/pipe_reader /playpen/bin/
+COPY verify.sh /playpen/bin
+
+ENV LD_LIBRARY_PATH=/usr/local/lib64:/usr/local/lib
+ENV C_INCLUDE_PATH=/usr/local/include
+
+CMD [ "/playpen/bin/mc_listener" ]
--- /dev/null
+// vim: ts=4 sw=4 noet:
+/*
+--------------------------------------------------------------------------------
+ Copyright (c) 2018-2019 AT&T Intellectual Property.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+--------------------------------------------------------------------------------
+*/
+
+/*
+ Mnemonic: pipe_reader.c
+ Abstract: Read a single pipe which was associated with a message type. This
+ programme is primarily for verification or example of how to use the
+ read1() function in the mc-listener library.
+
+ Date: 22 August 2019
+ Author: E. Scott Daniels
+*/
+
+#include <unistd.h>
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <time.h>
+#include <string.h>
+
+
+#include "mcl.h"
+
+//---- support -----------------------------------------------------------------------------
+
+static void bad_arg( char* what ) {
+ fprintf( stderr, "[ERR] option is unrecognised or isn't followed by meaningful data: %s\n", what );
+}
+
+static void usage( char* argv0 ) {
+ fprintf( stderr, "usage: %s [-d fifo-dir] [-e] [-m msg-type] [-s]\n", argv0 );
+ fprintf( stderr, " -d dir (default is /tmp/mcl/fifos)\n" );
+ fprintf( stderr, " -e extended headers expected in FIFO data\n" );
+ fprintf( stderr, " -m msg-type (default is 0)\n" );
+ fprintf( stderr, " -s stats only mode\n" );
+}
+
+//------------------------------------------------------------------------------------------
+int main( int argc, char** argv ) {
+ void* ctx; // the mc listener library context
+ char* dname = "/tmp/mcl/fifos"; // default directory where we open fifos
+ int pidx = 1; // parameter index
+ int error = 0;
+ int len;
+ int mtype = 0;
+ char buf[4096];
+ int flush_often = 0;
+ int long_hdrs = 0; // if -e is on command line, we expect long headers
+ int stats_only = 0;
+ char timestamp[MCL_TSTAMP_SIZE]; // we'll get the timestamp from this
+ long count = 0;
+ int blabber = 0;
+
+ while( pidx < argc && argv[pidx][0] == '-' ) { // simple argument parsing (-x or -x value)
+ switch( argv[pidx][1] ) {
+ case 'd':
+ if( pidx+1 < argc ) {
+ dname = strdup( argv[pidx+1] );
+ pidx++;
+ } else {
+ bad_arg( argv[pidx] );
+ error = 1;
+ }
+ break;
+
+ case 'e':
+ long_hdrs = 1;
+ break;
+
+ case 'f':
+ flush_often = 1;
+ break;
+
+ case 'm':
+ if( pidx+1 < argc ) {
+ mtype = atoi( argv[pidx+1] );
+ pidx++;
+ } else {
+ bad_arg( argv[pidx] );
+ error = 1;
+ }
+ break;
+
+ case 's':
+ stats_only = 1;
+ break;
+
+ case 'h':
+ case '?':
+ usage( argv[0] );
+ exit( 0 );
+ break;
+
+ default:
+ bad_arg( argv[pidx] );
+ error = 1;
+ break;
+ }
+
+ pidx++;
+ }
+
+ if( error ) {
+ usage( argv[0] );
+ exit( 1 );
+ }
+
+ ctx = mcl_mk_context( dname ); // initialise the library context
+ if( ctx == NULL ) {
+ fprintf( stderr, "[FAIL] couldn't initialise the mc listener library" );
+ exit( 1 );
+ }
+
+ while( 1 ) {
+ len = mcl_fifo_tsread1( ctx, mtype, buf, sizeof( buf ) -1, long_hdrs, timestamp );
+ if( len > 0 ) {
+ if( stats_only ) {
+ if( time( NULL ) > blabber ) {
+ fprintf( stdout, "[%d] %ld messages received\n", mtype, count );
+ blabber = time( NULL ) + 2;
+ }
+ } else {
+ buf[len] = 0;
+ fprintf( stdout, "[%d] ts=%s count=%ld len=%d msg=%s\n", mtype, timestamp, count, len, buf );
+ if( flush_often ) {
+ fflush( stdout );
+ }
+ }
+
+ count++;
+ }
+ }
+
+ fprintf( stderr, "[INFO] mc listener is finished.\n" );
+}
+
--- /dev/null
+// vim: ts=4 sw=4 noet:
+/*
+--------------------------------------------------------------------------------
+ Copyright (c) 2018-2019 AT&T Intellectual Property.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+--------------------------------------------------------------------------------
+*/
+
+/*
+ Mnemonic: sender.c
+ Abstract: Very simple test sender. Sends messages with a given delay between each.
+ The sender also uses epoll_wait() to ensure that any received messages
+ don't clog the queue.
+
+ Parms: The following positional parameters are recognised on the command line:
+ [listen_port [delay [stats-freq] [msg-type]]]]
+
+ Defaults:
+ listen_port 43086
+ delay (mu-sec) 1000000 (1 sec)
+ stats-freq 10
+ msg-type 0
+
+ Date: 1 April 2019
+*/
+
+#include <unistd.h>
+#include <errno.h>
+#include <string.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <sys/epoll.h>
+#include <time.h>
+
+#include <rmr/rmr.h>
+
+int main( int argc, char** argv ) {
+ void* mrc; //msg router context
+ struct epoll_event events[1]; // list of events to give to epoll
+ struct epoll_event epe; // event definition for event to listen to
+ int ep_fd = -1; // epoll's file des (given to epoll_wait)
+ int rcv_fd; // file des that NNG tickles -- give this to epoll to listen on
+ int nready; // number of events ready for receive
+ rmr_mbuf_t* sbuf; // send buffer
+ rmr_mbuf_t* rbuf; // received buffer
+ int count = 0;
+ int rcvd_count = 0;
+ char* listen_port = "43086";
+ int delay = 1000000; // mu-sec delay between messages
+ int mtype = 0;
+ int stats_freq = 100;
+
+ if( argc > 1 ) {
+ listen_port = argv[1];
+ }
+ if( argc > 2 ) {
+ delay = atoi( argv[2] );
+ }
+ if( argc > 3 ) {
+ mtype = atoi( argv[3] );
+ }
+
+ fprintf( stderr, "<DEMO> listen port: %s; mtype: %d; delay: %d\n", listen_port, mtype, delay );
+
+ if( (mrc = rmr_init( listen_port, 1400, RMRFL_NONE )) == NULL ) {
+ fprintf( stderr, "<DEMO> unable to initialise RMr\n" );
+ exit( 1 );
+ }
+
+ rcv_fd = rmr_get_rcvfd( mrc ); // set up epoll things, start by getting the FD from MRr
+ if( rcv_fd < 0 ) {
+ fprintf( stderr, "<DEMO> unable to set up polling fd\n" );
+ exit( 1 );
+ }
+ if( (ep_fd = epoll_create1( 0 )) < 0 ) {
+ fprintf( stderr, "[FAIL] unable to create epoll fd: %d\n", errno );
+ exit( 1 );
+ }
+ epe.events = EPOLLIN;
+ epe.data.fd = rcv_fd;
+
+ if( epoll_ctl( ep_fd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 ) {
+ fprintf( stderr, "[FAIL] epoll_ctl status not 0 : %s\n", strerror( errno ) );
+ exit( 1 );
+ }
+
+ sbuf = rmr_alloc_msg( mrc, 256 ); // alloc first send buffer; subsequent buffers allcoated on send
+ rbuf = NULL; // don't need to alloc receive buffer
+
+ while( ! rmr_ready( mrc ) ) { // must have a route table before we can send; wait til RMr say it has one
+ sleep( 1 );
+ }
+ fprintf( stderr, "<DEMO> rmr is ready\n" );
+
+
+ while( 1 ) { // send messages until the cows come home
+ snprintf( sbuf->payload, 200, "count=%d received= %d ts=%lld %d stand up and cheer!\n", // create the payload
+ count, rcvd_count, (long long) time( NULL ), rand() );
+
+ sbuf->mtype = mtype; // fill in the message bits
+ sbuf->len = strlen( sbuf->payload ) + 1; // our receiver likely wants a nice acsii-z string
+ sbuf->state = 0;
+ sbuf = rmr_send_msg( mrc, sbuf ); // send it (send returns an empty payload on success, or the original payload on fail/retry)
+ while( sbuf->state == RMR_ERR_RETRY ) { // soft failure (device busy?) retry
+ sbuf = rmr_send_msg( mrc, sbuf ); // retry send until it's good (simple test; real programmes should do better)
+ }
+ count++;
+ fprintf( stderr, "<SNDR> sent message\n" );
+
+ usleep( delay );
+ mtype++;
+ if( mtype > 6 ) {
+ mtype = 1;
+ }
+ }
+}
+
--- /dev/null
+// vim: ts=4 sw=4 noet:
+/*
+ --------------------------------------------------------------------------------
+ Copyright (c) 2018-2019 AT&T Intellectual Property.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ --------------------------------------------------------------------------------
+*/
+
+/*
+ Mnemonic: unit_test.c
+ Abstract: Basic unit tests for the mc listener.
+ Date: 22 August 2019
+ Author: E. Scott Daniels
+*/
+
+// this/these are what we are testing; include them directly
+#include "mcl.c"
+
+/*
+ Parms: [fifo-dir-name]
+*/
+int main( int argc, char** argv ) {
+ void* ctx;
+ int errors;
+ char* dname = "/tmp/fifos";
+ char* port = "4560";
+ int fd;
+ int fd2;
+
+ if( argc > 1 ) {
+ dname = argv[1];
+ }
+
+ ctx = mcl_mk_context( dname );
+ if( ctx == NULL ) {
+ fprintf( stderr, "[FAIL] couldn't make context" );
+ exit( 1 );
+ }
+
+ mcl_set_sigh(); // prevent colobber from broken pipe
+
+ fd = suss_fifo( ctx, 101, 1 ); // should open the file for writing and return the fdes
+ if( fd < 0 ) {
+ fprintf( stderr, "[FAIL] suss_fifo did not return a valid fd\n" );
+ errors++;
+ }
+
+ fd2= suss_fifo( ctx, 101, 0 ); // should open the file file for reading and return a different fd
+ if( fd < 0 ) {
+ fprintf( stderr, "[FAIL] suss_fifo did not return a valid fd\n" );
+ errors++;
+ }
+ if( fd == fd2 ) {
+ fprintf( stderr, "[FAIL] reading and writing fifo file descriptors expected to differ; both were %d\n", fd );
+ errors++;
+ }
+
+ mcl_start_listening( ctx, port, 0 ); // start the listener, no waiting for rt since we don't send
+
+ if( ! errors ) {
+ fprintf( stderr, "[PASS] all tests look peachy\n" );
+ }
+
+ return errors != 0;
+}
+
--- /dev/null
+#!/usr/bin/env bash
+# vim: ts=4 sw=4 noet:
+#----------------------------------------------------------------------------------
+#
+# Copyright (c) 2018-2019 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#---------------------------------------------------------------------------------
+
+
+# ----------------------------------------------------------------------
+# Mnemonic: verify.sh
+# Abstract: Simple script to attempt to verify that the mc_listener is
+# capable of running. This will start a listener and a sender
+# and then will cat a few lines from one of the FIFOs.
+# This script is designed to run using the geneated runtime
+# image; in other words, it expects to find the binaries
+# in /playpen/bin.
+#
+# Date: 26 August 2019
+# Author: E. Scott Daniels
+# ----------------------------------------------------------------------
+
+# run sender at a 2 msg/sec rate (500000 musec delay)
+# sender sends msg types 0-6 and the route table in /tmp
+# will direct them to the listener. We also need to switch
+# the RT listen port so there is no collision with the listen
+# preocess.
+#
+function run_sender {
+ echo "starting sender"
+ RMR_SEED_RT=/tmp/local.rt RMR_RTG_SVC=9989 /playpen/bin/sender 43086 10000 >/tmp/sender.log 2>&1 &
+ spid=$!
+ sleep 10
+
+ echo "stopping sender"
+ kill -15 $spid
+}
+
+function run_listener {
+ echo "starting listener"
+ /playpen/bin/mc_listener $ext_hdr -r 1 -d $fifo_dir >/tmp/listen.log 2>&1 &
+ lpid=$!
+
+ sleep 15
+ echo "stopping listener"
+ kill -15 $lpid
+}
+
+# run a pipe reader for one message type
+function run_pr {
+ echo "starting pipe reader $1"
+ /playpen/bin/pipe_reader $ext_hdr -m $1 -d $fifo_dir >/tmp/pr.$1.log 2>&1 &
+ #/playpen/bin/pipe_reader -m $1 -d $fifo_dir & # >/tmp/pr.$1.log 2>&1
+ typeset prpid=$!
+
+ sleep 12
+ echo "stopping pipe reader $1"
+ kill -1 $prpid
+}
+
+# generate a dummy route table that the sender needs
+function gen_rt {
+ cat <<endKat >/tmp/local.rt
+ newrt|start
+ mse | 0 | -1 | localhost:4560
+ mse | 1 | -1 | localhost:4560
+ mse | 2 | -1 | localhost:4560
+ mse | 3 | -1 | localhost:4560
+ mse | 4 | -1 | localhost:4560
+ mse | 5 | -1 | localhost:4560
+ mse | 6 | -1 | localhost:4560
+ newrt|end
+endKat
+}
+
+# ---- run everything ---------------------------------------------------
+
+ext_hdr="-e" # run with extended header enabled
+fifo_dir=/tmp/fifos
+mkdir -p $fifo_dir # redirect fifos so we don't depend on mount
+
+gen_rt # generate a dummy route table
+run_listener &
+sleep 4
+
+for p in 0 1 2 3 4 5 6
+do
+ run_pr $p &
+done
+sleep 1
+run_sender &
+
+sleep 20 # long enough for all functions to finish w/o having to risk a wait hanging
+echo "all functions stopped; looking at logs"
+
+# ---------- validation -------------------------------------------------
+
+errors=0
+
+# logs should be > 0 in size
+echo "----- logs ---------"
+ls -al /tmp/*.log
+
+# pipe reader log files 1-6 should have 'stand up and cheer' messages
+# pipe reader log for MT 0 will likley be empty as sender sends only
+# one of those and buffer not likely flushed. So, we only check 1-6
+#
+for l in 1 2 3 4 5 6
+do
+ if [[ ! -s /tmp/pr.$l.log ]]
+ then
+ echo "[FAIL] log $l was empty"
+ (( errors++ ))
+ else
+ if ! grep -q -i "stand up and cheer" /tmp/pr.$l.log
+ then
+ echo "[FAIL] pipe reader log did not have any valid messages: /tmp/pr.$l.log"
+ (( errors++ ))
+ fi
+ fi
+done
+
+if (( ! errors ))
+then
+ echo "[OK] All logs seem good"
+fi
+
+nfifos=$( ls /tmp/fifos/MT_* | wc -l )
+if (( nfifos < 7 ))
+then
+ echo "didn't find enough fifos"
+ ls -al /tmp/fifos/*
+ (( errors++ ))
+else
+ echo "[OK] Found expected fifos"
+fi
+
+if (( errors ))
+then
+ echo "[FAIL] $errors errors noticed"
+else
+ echo "[PASS]"
+fi
+
Code Review / ric-app / mc.git / commitdiff