Skip to content

Instantly share code, notes, and snippets.

@binduchinnasamy
Created June 26, 2018 08:18
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save binduchinnasamy/a1b9f424505d7063bb28aace19870611 to your computer and use it in GitHub Desktop.
Read IBM MQ message from Azure Service Fabric Stateless service
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Fabric;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.ServiceBus.Messaging;
using Microsoft.ServiceFabric.Services.Communication.Runtime;
using Microsoft.ServiceFabric.Services.Runtime;
using Reader.ServiceListeners;
using IBM.WMQ;
using System.Collections;
namespace Reader
{
/// <summary>
/// An instance of this class is created for each service instance by the Service Fabric runtime.
/// </summary>
internal sealed class Reader : StatelessService
{
private const int READ_OPTIONS = MQC.MQOO_OUTPUT + MQC.MQOO_FAIL_IF_QUIESCING + MQC.MQOO_INQUIRE + MQC.MQOO_INPUT_AS_Q_DEF;
private const int WRITE_OPTIONS = MQC.MQOO_OUTPUT + MQC.MQOO_FAIL_IF_QUIESCING + MQC.MQOO_SET_ALL_CONTEXT + MQC.MQPMO_PASS_IDENTITY_CONTEXT + MQC.MQPMO_PASS_ALL_CONTEXT;
private string _qManagerName;
private string _channelName;
private string _connectionDetails;
private string _userName;
private string _password;
private string _qName;
public Reader(StatelessServiceContext context)
: base(context)
{
//Get MQ Connection details from Service Fabric Configuration
var config = Context.CodePackageActivationContext.GetConfigurationPackageObject("Config");
_qManagerName = config.Settings.Sections["MQConnection"].Parameters["Q_MANAGER_NAME"].Value;
_channelName = config.Settings.Sections["MQConnection"].Parameters["CHANNEL_NAME"].Value;
_connectionDetails = config.Settings.Sections["MQConnection"].Parameters["CONNECTION_DETAILS"].Value;
_userName = config.Settings.Sections["MQConnection"].Parameters["USER_NAME"].Value;
_password = config.Settings.Sections["MQConnection"].Parameters["PASSWORD"].Value;
_qName = config.Settings.Sections["MQConnection"].Parameters["Q_NAME"].Value;
}
/// <summary>
/// Optional override to create listeners (e.g., TCP, HTTP) for this service replica to handle client or user requests.
/// </summary>
/// <returns>A collection of listeners.</returns>
protected override IEnumerable<ServiceInstanceListener> CreateServiceInstanceListeners()
{
return new ServiceInstanceListener[0];
}
/// <summary>
/// This is the main entry point for your service instance.
/// </summary>
/// <param name="cancellationToken">Canceled when Service Fabric needs to shut down this service instance.</param>
protected override async Task RunAsync(CancellationToken cancellationToken)
{
//_queueManager = new MQQueueManager(configuration.QueueManagerName, configuration.ChannelName, configuration.ServerName);
Hashtable mqProps = new Hashtable();
mqProps.Add(MQC.CONNECTION_NAME_PROPERTY, _connectionDetails);
mqProps.Add(MQC.CHANNEL_PROPERTY, _channelName);
mqProps.Add(MQC.USER_ID_PROPERTY, _userName);
mqProps.Add(MQC.PASSWORD_PROPERTY, _password);
MQQueue queue;
try
{
MQQueueManager mqQMgr = new MQQueueManager(_qManagerName, _channelName, _connectionDetails);
mqQMgr.Connect();
queue = mqQMgr.AccessQueue(_qName, READ_OPTIONS);
}
catch (Exception Oex)
{
ServiceEventSource.Current.ServiceMessage(this.Context, Oex.StackTrace);
}
ServiceEventSource.Current.ServiceMessage(this.Context, "Connected to MQ Server");
while (true)
{
cancellationToken.ThrowIfCancellationRequested();
//READ MESSAGE
MQMessage readQueueMessage = new MQMessage();
//readQueueMessage.Format = MQC.MQFMT_STRING;
MQGetMessageOptions queueGetMessageOptions = new MQGetMessageOptions();
try
{
//queue.Get(readQueueMessage, queueGetMessageOptions);
string readMessage = readQueueMessage.ReadString(readQueueMessage.MessageLength);
ServiceEventSource.Current.ServiceMessage(this.Context, "Message Picked up by the reader : Read Queue Message is :" + readMessage);
}
catch (MQException exp)
{
if(exp.Message == "2033")
{
//Do nothing, false alarm - as the queue is empty
//ServiceEventSource.Current.ServiceMessage(this.Context, "The queue is empty");
}
else
{
ServiceEventSource.Current.ServiceMessage(this.Context, "Exception: " + exp.Message);
}
}
await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken);
}
}
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment