ar429_UDPCommRxEmbedded.c

/*******************************************************************************
//
//   File:
//       ar429_UDPCommRxEmbedded.c
//
//       Copyright (c) MAX Technologies Inc. 1988-2017, All Rights Reserved.
//             CONFIDENTIAL AND PROPRIETARY INFORMATION WHICH IS THE 
//                         PROPERTY OF MAX TECHNOLOGIES INC. 
//
//    This code is to be compiled under Linux to create the file ar429_UDPCommRxEmbedded.mxf.
//    It must then be transferred (by tftp or any mean) on the FM in the maxfiles folder to be used.
//    After being started, it will wait for a UDP datagram sent by ar429_UDPCommTX to come.
//    Once it is received, it will create A429 records from it, start an aperiodic transmission on the 
//    1st A429 TX channel and receive it on the corresponding RX channel through internal LB.
//    The records are then analyzed, another UDP datagram is created from it, and sent back to the host.
//
//    Hardware Requirements:
//       - MAXT FlexMulti with AR429 ports on it
//
*******************************************************************************/

#include "example.h"

#define BUFFER_SIZE  100
#define NUM_REC      4

void udpInit(void);
void udpClose(int socket);

int main(void)
{
   HMXF_SERVER server =0;
   HMXF_CHANNEL txChannel=0, rxChannel=0;
   HMXF_BUFFER  rxBuffer=0,txBuffer=0;
   MXF_A429_DATAREC *txRec429=0, *rxRec429=0,*rec, *recPtr;
   struct sockaddr_in rxPort;
   int                rxPortSocket=0;
   char receivedStrg[BUFFER_SIZE] ="",send_buffer[BUFFER_SIZE]="";
   char temp_buffer[BUFFER_SIZE]="";
   char * pEnd;
   uint64 i=0,j=0,record=0, parity=0, sdi=0, ssm=0, status=0, bytesCount=0;
   uint64 delay100ms = 100000, msgCount=0, usedBytes=0, freeBytes=0;
   int labelArray[NUM_REC], dataRecArray[NUM_REC];
   uint64 labelArrayUint64[NUM_REC], dataRecArrayUint64[NUM_REC];
   uint32 rc=0;
   int bytes_received=0,bytes_sent=0;
   size_t client_length=0;

   memset(labelArray, 0, sizeof(labelArray));
   memset(dataRecArray, 0, sizeof(dataRecArray));
   memset(labelArrayUint64, 0, sizeof(labelArrayUint64));
   memset(dataRecArrayUint64, 0, sizeof(dataRecArrayUint64));

   //MXF section
   rc = mxfServerConnect("0.0.0.0", "", "", FALSE, &server);

   if(!rc)
      rc = mxfSystemInit(server);

   // Set timebase to 64-bit microseconds
   if (!rc)
      rc = mxfSystemTimeBaseSet(server, MXF_TIMEBASE_DEVICE_USEC);

   if(!rc)
      rc = mxfChannelGet(server, MXF_CLASS_A429, MXF_SCLASS_TX_CHANNEL, MXF_MODULE_A429_EH, 0, &txChannel);
   if(!rc)
      rc = mxfChannelGet(server, MXF_CLASS_A429, MXF_SCLASS_RX_CHANNEL, MXF_MODULE_A429_EH, 0, &rxChannel);

   if(!rc)
      rc = mxfAttributeUint64Set(rxChannel, KMXF_A429_TX_RX_TEST_LB, TRUE);

   // Allocate RX acquisition buffer
   if (!rc)
      rc = mxfRxAcqBufferAlloc(rxChannel, BUFFER_SIZE, &rxBuffer, NULL);

   // Allocate Aperiodic TX static buffer for HIGH priority queue
   if (!rc)
      rc = mxfTxAperiodicBufferAlloc(txChannel, MXF_TXAPERIODIC_PRIORITY_HIGH, BUFFER_SIZE, &txBuffer, NULL);

   // Allocate host buffer
   if (!rc)
   {
      txRec429 = (MXF_A429_DATAREC*)malloc(BUFFER_SIZE);
      if (!txRec429)
         rc = MAXT_ERROR_MEM;
   }

   if (!rc)
   {
      rxRec429 = (MXF_A429_DATAREC*)malloc(BUFFER_SIZE);
      if (!rxRec429)
         rc = MAXT_ERROR_MEM;
   }

   // Start the acquisition process
   if (!rc)
      rc = mxfRxAcqModeSet(rxBuffer, MXF_RXACQ_MODE_LINEAR);
   if (!rc)
   {
      rc = mxfRxAcqStart(rxBuffer, MXF_RXACQ_FLAG_DEFAULT, 0, 0);
      if (!rc)
         printf("\nAcquisition started\n\r");
   }

   //UDP section
   if(!rc)
   {
      /* Clear out server struct */
      memset((void*)&rxPort, '\0', sizeof(struct sockaddr_in));

      client_length = sizeof(struct sockaddr_in);

      rxPort.sin_family = AF_INET;
      rxPort.sin_addr.s_addr = htonl(INADDR_ANY);
      rxPort.sin_port = htons(8888); // port number to choose

      udpInit();

      //Creates the socket
      rxPortSocket = (int)socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
      if(rxPortSocket == -1)
      {
         printf("Could not create socket.\n");
      }

      //Binds the socket
      if(!rc)
         rc = bind(rxPortSocket, (struct sockaddr*)&rxPort, sizeof(struct sockaddr_in));
      if(rc!=0)
      {
         printf("Could not bind name to socket.\n");
         udpClose(rxPortSocket);
      }

      printf("Waiting for datagram...\n");
      bytes_received = recvfrom(rxPortSocket, receivedStrg, BUFFER_SIZE, 0, (struct sockaddr*)&(rxPort), (socklen_t*)&client_length);

      if(bytes_received != -1)
      {
         printf("Datagram received.\n");
         i=0;
         do
         {
            //Treats the labels
            if(receivedStrg[i]=='L')
            {
               j=0;
               pEnd = &receivedStrg[i+1];
               while(pEnd[1] != 'D')
               {
                  labelArray[j] = (int)(strtol(pEnd, &pEnd, 10));
                  j++;
               }
            }

            //Treats the data
            if(receivedStrg[i]=='D')
            {
               j=0;
               pEnd = &receivedStrg[i+1];
               while(pEnd[1] != '\0')
               {
                  dataRecArray[j] = (int)(strtol(pEnd, &pEnd, 10));
                  j++;
               }
            }
            i++;
         } while(receivedStrg[i] != '\0');
      }
   }

   //MXF section
   // Set the record
   rec = txRec429;
   for (record=0; record<NUM_REC && !rc; record++)
   {
      // Set each record: ARINC 429 SDI=1
      rec->timeTag = 0;
      rec->control = 0;
      rec->repeatCount = 1;
      rec->reserved = 0;

      sdi = 1;
      ssm = 0;
      parity = VMXF_A429_PARITY_ODD;
      rc = mxfA429ArwCompose(labelArray[record], sdi, dataRecArray[record], ssm, parity, &rec->data);

      if(!rc)
         rc = mxfA429NextDataRecordPtrGet(rec, &rec);
   }

   // Transmit the array of records
   if(!rc)
      rc = mxfA429TxAperiodicWrite(txBuffer, MXF_TXAPERIODIC_FLAG_DEFAULT, delay100ms, NUM_REC, txRec429);

   // Wait till all records are sent on the interface
   do
   {
      if(!rc)
         rc = mxfTxAperiodicBufferStatusGet(txBuffer, &msgCount, &usedBytes, &freeBytes);
   }while (msgCount!=0 && !rc);

   // Make sure they are received
   mxfSleep(100);

   // Read and display records
   if(!rc)
      rc = mxfA429RxAcqRead(rxBuffer, 0, BUFFER_SIZE, &status, &msgCount, &bytesCount, rxRec429);

   recPtr = rxRec429;

   for (record=0; record<msgCount && !rc; record++)
   {
      rc = mxfA429ArwDecompose(recPtr->data, &labelArrayUint64[record], &sdi, &dataRecArrayUint64[record], &ssm, &parity);

      if(!rc)
         rc = mxfA429NextDataRecordPtrGet(recPtr, &recPtr);
   }

   if(!rc)
   {
      //Creation of the string to send back to the host
      strcat(send_buffer, "L");
      for(i=0; i<NUM_REC; i++)
      {
         sprintf(temp_buffer, " %llu", labelArrayUint64[i]);
         strcat(send_buffer, temp_buffer);
      }

      strcat(send_buffer, " D");
      for(i=0; i<NUM_REC; i++)
      {
         sprintf(temp_buffer, " %llu", dataRecArrayUint64[i]);
         strcat(send_buffer, temp_buffer);
      }

      //UDP section
      /* Send data back */
      printf("Sending datagram...\n");
      client_length = sizeof(struct sockaddr_in);
      bytes_sent = sendto(rxPortSocket, send_buffer, (int)strlen(send_buffer) + 1, 0, (struct sockaddr*)&rxPort, (int)client_length);

      if(bytes_sent == -1)
      {
         printf("Error sending datagram.\n");
         udpClose(rxPortSocket);
      }
      else
         printf("Datagram sent successfully.\n");
   }
   if(rxPortSocket)
      udpClose(rxPortSocket);

   //MXF Section
   // Stop and flush unread data
   if (!rc)
      rc = mxfRxAcqStop(rxBuffer);
   if (!rc)
   {
      rc = mxfRxAcqClear(rxBuffer);
      if (!rc)
         printf("\nAcquisition stopped\n\r");
   }

   // Catch any previous error
   if (rc)
   {
      char buffer[256];
      rc = mxfSystemErrorStringGet(server, rc, sizeof(buffer), buffer);
      printf("%s\n", buffer);
   }

   printf("\nTerminating\n");

   mxfSystemTerminate(server);
   mxfServerDisconnect(server);

   printf("\n\nPress a key to terminate\n");
   getchar();

   return rc;
}

void udpInit(void)
{
#ifdef _linux
#else
   WSADATA wsaData = { 0 };
   int iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);

   if (iResult != 0)
   {
      wprintf(L"WSAStartup failed: %d\n", iResult);
      return;
   }
#endif
}

void udpClose(int socket)
{
#ifdef _linux
   shutdown(socket, SHUT_RDWR);
   close(socket);
#else
   closesocket(socket);
   WSACleanup();
#endif
}