myfix.cpp File Reference

#include <iostream>
#include <memory>
#include <fstream>
#include <iomanip>
#include <sstream>
#include <vector>
#include <map>
#include <list>
#include <set>
#include <iterator>
#include <algorithm>
#include <typeinfo>
#include <thread>
#include <sys/ioctl.h>
#include <signal.h>
#include <termios.h>
#include <errno.h>
#include <string.h>
#include <fix8/f8includes.hpp>
#include <fix8/usage.hpp>
#include <fix8/consolemenu.hpp>
#include <fix8/multisession.hpp>
#include "Myfix_types.hpp"
#include "Myfix_router.hpp"
#include "Myfix_classes.hpp"
#include "myfix.hpp"

Go to the source code of this file.

Namespaces
	FIX8
	FIX8::TEX

Functions
void	print_usage ()
const string	GETARGLIST ("hl:svqc:R:S:rdomN:D:")
f8_atomic< bool >	term_received (false)
void	server_process (ServerSessionBase *srv, int scnt, bool ismulti=false)
void	client_process (ClientSessionBase *mc)
bool	quiet (false)
unsigned	next_send (0)
unsigned	next_receive (0)
void	sig_handler (int sig)
int	main (int argc, char **argv)

Variables
FIX8::tty_save_state	save_tty (0)

Detailed Description

This is a complete working example of a FIX client/server using FIX8.

Usage: f8test [-NRScdhlmoqrsv]
-N,–session for client, select session to use from configuration (default none)
-R,–receive set next expected receive sequence number
-S,–send set next send sequence number
-c,–config xml config (default: myfix_client.xml or myfix_server.xml)
-d,–dump dump parsed XML config file, exit
-h,–help help, this screen
-l,–log global log filename
-m,–multi run multiple server mode (default single server session at a time)
-o,–once for server, allow one client session then exit
-q,–quiet do not print fix output
-r,–reliable start in reliable mode
-s,–server run in server mode (default client mode)
-v,–version print version, exit


To use start the server:

% f8test -sl server

In another terminal session, start the client:

% f8test -l client

Notes

1. The client has a simple menu. Press ? to see options.
   
2. The server will wait for the client to logout before exiting.
   
3. The server uses myfix_client.xml and the client uses myfix_server.xml for configuration settings.
   
4. The example uses the files FIX50SP2.xml and FIXT11.xml in ./schema
   
   

Definition in file myfix.cpp.

Function Documentation

void client_process

(

ClientSessionBase *

mc

)

Definition at line 376 of file myfix.cpp.

References FIX8::ClientSessionBase::has_given_up(), FIX8::Session::is_shutdown(), save_tty, FIX8::ClientSessionBase::session_ptr(), FIX8::Session::stop(), and term_received().

Referenced by main().

{
   MyMenu mymenu(*mc->session_ptr(), 0, cout);
   cout << endl << "Menu started. Press '?' for help..." << endl << endl;
   char ch(0);
   mymenu.get_tty().set_raw_mode();
   save_tty = mymenu.get_tty();
   while(!mymenu.get_istr().get(ch).bad() && !mc->has_given_up() && !term_received && ch != 0x3 && mymenu.process(ch))
      ;
   // don't explicitly call mc->session_ptr()->stop() with reliable sessions
   // before checking if the session is already shutdown - the framework will generally do this for you
   if (!mc->session_ptr()->is_shutdown())
      mc->session_ptr()->stop();

   mymenu.get_tty().unset_raw_mode();
}

const string GETARGLIST

(

"hl:svqc:R:S:rdomN:D:"

)

Referenced by main(), and print_usage().

int main	(	int	argc,
		char **	argv
	)

Definition at line 181 of file myfix.cpp.

References FIX8::SessionConfig::_session_name, FIX8::SessionManager< T >::add(), client_process(), FIX8::TEX::ctx(), XmlElement::Factory(), FIX8_PACKAGE, FIX8_VERSION, FIX8::SessionManager< T >::for_each_if(), GETARGLIST(), RandDev::getrandom(), glout_error, FIX8::hypersleep< h_seconds >(), RandDev::init(), next_receive(), next_send(), print_usage(), quiet(), save_tty, server_process(), FIX8::ClientSessionBase::session_ptr(), sig_handler(), term_received(), FIX8::tty_save_state::unset_raw_mode(), and FIX8::f8Exception::what().

{
   int val;
   bool server(false), reliable(false), once(false), dump(false), multi(false);
   string clcf, session;

#ifdef FIX8_HAVE_GETOPT_LONG
   option long_options[]
   {
      { "help",      0, 0, 'h' },
      { "version",   0, 0, 'v' },
      { "log",       1, 0, 'l' },
      { "delimiter", 1, 0, 'D' },
      { "config",    1, 0, 'c' },
      { "session",   1, 0, 'N' },
      { "once",      0, 0, 'o' },
      { "server",    0, 0, 's' },
      { "multi",     0, 0, 'm' },
      { "send",      1, 0, 'S' },
      { "receive",   1, 0, 'R' },
      { "quiet",     0, 0, 'q' },
      { "reliable",  0, 0, 'r' },
      { "dump",      0, 0, 'd' },
      { 0 },
   };

   while ((val = getopt_long (argc, argv, GETARGLIST.c_str(), long_options, 0)) != -1)
#else
   while ((val = getopt (argc, argv, GETARGLIST.c_str())) != -1)
#endif
   {
      switch (val)
      {
      case 'v':
         cout << argv[0] << " for " FIX8_PACKAGE " version " FIX8_VERSION << endl;
         cout << "Released under the GNU LESSER GENERAL PUBLIC LICENSE, Version 3. See <http://fsf.org/> for details." << endl;
         return 0;
      case ':': case '?': return 1;
      case 'h': print_usage(); return 0;
      case 'l': GlobalLogger::set_global_filename(optarg); break;
      case 'D': GlobalLogger::set_delimiter(optarg); break;
      case 'c': clcf = optarg; break;
      case 's': server = true; break;
      case 'N': session = optarg; break;
      case 'm': multi = true; break;
      case 'o': once = true; break;
      case 'S': next_send = stoul(optarg); break;
      case 'R': next_receive = stoul(optarg); break;
      case 'q': quiet = true; break;
      case 'r': reliable = true; break;
      case 'd': dump = true; break;
      default: break;
      }
   }

   RandDev::init();

   signal(SIGTERM, sig_handler);
   signal(SIGINT, sig_handler);
#ifndef _MSC_VER
   signal(SIGQUIT, sig_handler);
#endif

   bool restore_tty(false);

   try
   {
      const string conf_file(server ? clcf.empty() ? "myfix_server.xml" : clcf : clcf.empty() ? "myfix_client.xml" : clcf);
      f8_atomic<unsigned> scnt(0);

      if (dump)
      {
         XmlElement *root(XmlElement::Factory(conf_file));
         if (root)
            cout << *root << endl;
         else
            cerr << "Failed to parse " << conf_file << endl;
         return 0;
      }

      if (server)
      {
         if (multi)  // demonstrate use of multi session server manager
         {
            unique_ptr<ServerManager> sm(new ServerManager);
            sm->add(new ServerSession<myfix_session_server>(TEX::ctx(), conf_file, "TEX1"));
            sm->add(new ServerSession<myfix_session_server>(TEX::ctx(), conf_file, "TEX2"));

            vector<thread> thrds;
            while (!term_received)
            {
               ServerSessionBase *srv(sm->select());
               if (srv)
               {
                  thrds.push_back(thread ([&]() { server_process(srv, ++scnt, true); }));
                  hypersleep<h_seconds>(1);
               }
            }
            for_each(thrds.begin(), thrds.end(), [](thread& tt) { if (tt.joinable()) tt.join(); });
         }
         else  // serial server instances only
         {
            unique_ptr<ServerSessionBase> srv(new ServerSession<myfix_session_server>(TEX::ctx(), conf_file, "TEX"));

            while (!term_received)
            {
               if (!srv->poll())
                  continue;
               server_process(srv.get(), ++scnt);
               if (once)
                  break;
            }
         }
      }
      else
      {
         if (multi)  // demonstrate use of multi session client manager
         {
            const f8String cl1("DLD1"), cl2("DLD2");
            ClientManager cm;
            cm.add(cl1, reliable ? new ReliableClientSession<myfix_session_client>(TEX::ctx(), conf_file, cl1)
                                 : new ClientSession<myfix_session_client>(TEX::ctx(), conf_file, cl1));
            cm.add(cl2, reliable ? new ReliableClientSession<myfix_session_client>(TEX::ctx(), conf_file, cl2)
                                 : new ClientSession<myfix_session_client>(TEX::ctx(), conf_file, cl2));

            ClientSessionBase *csb(cm.for_each_if([&](ClientSessionBase *pp)
            {
               if (!quiet)
                  pp->session_ptr()->control() |= Session::printnohb;
               pp->start(false, next_send, next_receive, pp->session_ptr()->get_login_parameters()._davi());
               // if you use ReliableClientSession, you need to return true here always
               return reliable ? true : States::is_live(pp->session_ptr()->get_session_state());
            }));

            if (csb)
               cerr << csb->_session_name << " failed to start" << endl;

            vector<thread> thrds;
            cm.for_each_if([&](ClientSessionBase *pp)
            {
               // we should be running client_process however this won't work
               // since two threads can't share the same console
               thrds.push_back(thread ([=]()    // use copy closure
               {
                  MyMenu mymenu(*pp->session_ptr(), 0, cout);
                  hypersleep<h_seconds>(1 + RandDev::getrandom(10));
                  mymenu.new_order_single(); // send an order and then logout
                  mymenu.do_logout();
               }));
               return true;
            });

            for_each(thrds.begin(), thrds.end(), [](thread& tt) { if (tt.joinable()) tt.join(); });
         }
         else  // single client only
         {
            const string my_session(session.empty() ? "DLD1" : session);
            unique_ptr<ClientSessionBase>
               mc(reliable ? new ReliableClientSession<myfix_session_client>(TEX::ctx(), conf_file, my_session)
                           : new ClientSession<myfix_session_client>(TEX::ctx(), conf_file, my_session));
            if (!quiet)
               mc->session_ptr()->control() |= Session::printnohb;
            mc->start(false, next_send, next_receive, mc->session_ptr()->get_login_parameters()._davi());
            client_process(mc.get());
         }
      }
   }
   catch (f8Exception& e)
   {
      cerr << "exception: " << e.what() << endl;
      restore_tty = true;
      glout_error << e.what();
   }
   catch (exception& e) // also catches Poco::Net::NetException
   {
      cerr << "std::exception: " << e.what() << endl;
      restore_tty = true;
      glout_error << e.what();
   }
   catch (...)
   {
      cerr << "unknown exception" << endl;
      restore_tty = true;
      glout_error << "unknown exception";
   }

   if (restore_tty && !server)
      save_tty.unset_raw_mode();

   if (term_received)
      cout << endl << "terminated." << endl;
   return 0;
}

unsigned next_receive

(

0

)

Referenced by main(), and server_process().

unsigned next_send

(

0

)

Referenced by main(), and server_process().

void print_usage

(

)

Definition at line 744 of file myfix.cpp.

References UsageMan::add(), GETARGLIST(), UsageMan::print(), and UsageMan::setdesc().

Referenced by main().

{
   UsageMan um("f8test", GETARGLIST, "");
   um.setdesc("f8test -- f8 test client/server");
   um.add('s', "server", "run in server mode (default client mode)");
   um.add('m', "multi", "run in multiple server or client mode (default serial server or single client session at a time)");
   um.add('h', "help", "help, this screen");
   um.add('v', "version", "print version, exit");
   um.add('l', "log", "global log filename");
   um.add('o', "once", "for server, allow one client session then exit");
   um.add('c', "config", "xml config (default: myfix_client.xml or myfix_server.xml)");
   um.add('q', "quiet", "do not print fix output");
   um.add('R', "receive", "set next expected receive sequence number");
   um.add('S', "send", "set next send sequence number");
   um.add('D', "delimiter", "set GlobalLogger field delimiter (default ' ')");
   um.add('N', "session", "for client, select session to use from configuration (default DLD1)");
   um.add('r', "reliable", "start in reliable mode");
   um.add('d', "dump", "dump parsed XML config file, exit");
   um.add("e.g.");
   um.add("@f8test -sml server_log");
   um.add("@f8test -rl client_log");
   um.add("@f8test -rl server -S 124");
   um.print(cerr);
}

bool quiet

(

false

)

Referenced by main(), and server_process().

void server_process	(	ServerSessionBase *	srv,
		int	scnt,
		bool	ismulti = false
	)

Definition at line 394 of file myfix.cpp.

References FIX8::SessionConfig::_ses, FIX8::ServerSessionBase::create_server_instance(), FIX8::Configuration::get_process_model(), glout_info, FIX8::hypersleep< h_milliseconds >(), next_receive(), next_send(), FIX8::pm_pipeline, quiet(), and myfix_session_server::sample_scheduler_callback().

Referenced by main().

{
   unique_ptr<SessionInstanceBase> inst(srv->create_server_instance());
   if (!quiet)
      inst->session_ptr()->control() |= Session::print;
   glout_info << "client(" << scnt << ") connection established.";
   const ProcessModel pm(srv->get_process_model(srv->_ses));
   cout << (pm == pm_pipeline ? "Pipelined" : "Threaded") << " mode." << endl;
   inst->start(pm == pm_pipeline, next_send, next_receive);
   if (inst->session_ptr()->get_connection()->is_secure())
      cout << "Session is secure (SSL)" << endl;
   if (!ismulti && !quiet) // demonstrate use of timer events
   {
      TimerEvent<FIX8::Session> sample_callback(static_cast<bool (FIX8::Session::*)()>(&myfix_session_server::sample_scheduler_callback), true);
      inst->session_ptr()->get_timer().schedule(sample_callback, 60000); // call sample_scheduler_callback every minute forever
   }
   if (pm != pm_pipeline)
      while (!inst->session_ptr()->is_shutdown())
         FIX8::hypersleep<h_milliseconds>(100);
   cout << "Session(" << scnt << ") finished." << endl;
   inst->stop();
}

void sig_handler

(

int

sig

)

Definition at line 165 of file myfix.cpp.

References term_received().

Referenced by main().

{
   switch (sig)
   {
   case SIGTERM:
   case SIGINT:
#ifndef _MSC_VER
   case SIGQUIT:
#endif
      term_received = true;
      signal(sig, sig_handler);
      break;
   }
}

f8_atomic<bool> term_received

(

false

)

Referenced by client_process(), main(), and sig_handler().

Variable Documentation

FIX8::tty_save_state save_tty(0)

Referenced by client_process(), and main().