ProtoBuf-net: A helper method

The current implementation of protobuf-net does not contain a simple way of writing the proto as a string, or key-value pairs (unlike the C++ library).

Here’s a simple extension method to aid in logging protos, in case you ever have a similar need.

/// 
/// Returns the contents of the proto object log-friendly string format.
/// This method will also cycle through the repeating groups.
/// Eg: MyProto { fielda [100] = Foo, fieldb [101] = bar }
/// 
public static string ToDebug(this IExtensible proto) {
	StringBuilder sb = new StringBuilder();
	sb.AppendFormat("{0} {{", proto.GetType().Name);
	ReadProto(proto, sb);
	sb.Append("}");
	return sb.ToString();
}

private static void ReadProto(IExtensible proto, StringBuilder sb) {
	object val;
	SortedDictionary props = GetProtoAttributeProperties(proto);
	int[] keys = new int[props.Count];
	PropertyInfo info = null;
	MethodInfo mi = null;

	props.Keys.CopyTo(keys, 0);
	for (int i = 0; i < props.Keys.Count; i++) {
		info = props[keys[i]];
		val = info.GetValue(proto, null);
		if (val == null)
			continue;

		mi = proto.GetType().GetMethod(
					string.Format("ShouldSerialize{0}", props[keys[i]].Name), 
					BindingFlags.NonPublic | BindingFlags.Instance);
		if (mi != null && !Convert.ToBoolean(mi.Invoke(proto, null)))
			continue;
		sb.AppendFormat("{0} [{1}] = {2}",
			info.Name, keys[i], 
			val == null ? string.Empty :
				val is IExtensible ? val.GetType().Name : val.ToString());

		if (val is IExtensible) {
			sb.Append(" {");
			ReadProto((IExtensible)val, sb);
			sb.Append("}");
		} else {
			IList list = val as IList;
			if (list != null) {
				ReadRepeatingGroups(list, sb);
			} else if (i != keys.Length - 1)
				sb.Append(", ");
		} 
	}
}

private static void ReadRepeatingGroups(IList list, StringBuilder sb) {
	if (list == null)
		return;
	if (list.Count == 0) {
		sb.Append("{}");
		return;
	}
	sb.Append(" {");
	for (int i = 0; i < list.Count; i++) {
		sb.Append(" {");
		ReadProto((IExtensible)list[i], sb);
		sb.Append("}");
		if (i != list.Count - 1)
			sb.Append(", ");
	}
	sb.Append("}");
}

private static SortedDictionary GetProtoAttributeProperties(IExtensible proto) {
	SortedDictionary props = new SortedDictionary();
	PropertyInfo[] properties = proto.GetType().GetProperties();
	foreach (PropertyInfo pi in properties) {
		if (!pi.IsDefined(typeof(ProtoBuf.ProtoMemberAttribute), false))
			continue;
		ProtoMemberAttribute attr = null;
		object[] attributes = pi.GetCustomAttributes(false);
		foreach (object a in attributes) {
			attr = a as ProtoMemberAttribute;
			if (attr != null)
				break;
		}
		if (attr != null)
			props.Add(attr.Tag, pi);
	}
	return props;
}

QPID.net oddities

I noticed a couple of odd behavior in the QPID.net implementation so figured I’d document them here.

First, the code…

_client = new Client();
_client.connect( host, port, virtualHost, userName, password );
_session = _client.createSession( -1 );

// Create a response queue so the server can send us responses
_responseQueue = string.Format( "GUI_{0}_{1}", Environment.UserName, DateTime.Now.Ticks );
// Use the name of the response queue as the routing key
_session.queueDeclare( _responseQueue, new Option[] {
	// Setting EXCLUSIVE option makes the queue hang around
	// even when the session is closed
	Option.AUTO_DELETE //, Option.EXCLUSIVE
} );
_session.exchangeBind( _responseQueue, "amq.direct", _responseQueue );

1. Apparently, specifying Option.Exclusive on the queue, as the comment indicates, causes the queue to hang around even after you cleanly tear down the session and connection. (You can use qpid-stat to find out about the open sessions;
   qpid-stat -q localhost:6001 -S queue
   with the port where your daemon is listening to, of course.)

2. Even though I’m not specifying a timeout when creating the session, and given that QPID sessions typically are set to infinite timeouts, the session is closed by the broker after some period of inactivity. You’ll notice that the session has closed the next time you attempt to send a message back to the server. (If you have QPID configured to log everything, you can easily search the log file for your IP address to see if it’s disconnected; you should see entries similar to the following)
   debug DISCONNECTED [10.1.223.77:3356]

   For now, until I find a better solution, to fix this requires a hack. Essentially, you need to check whether the session is closed or not before attempting to send a message (code below). You’ll notice a Sleep call which I found was necessary as sending a message immediately after the session connects causes it to go into the ether

///
/// HACK! Attempt to reconnect if the session was closed.
/// TODO Find a way to keep the session open
///
protected void CheckSessionState() {
	if (_session != null && _session.Closed == false)
		return;
	bool isConnected = false;
	int tryCnt = 5, i = 0;
	while (!isConnected && i < tryCnt) {
		OnMessageWorkerLogMessage(this.GetServiceName() + " was closed, attempting to reconnect");
		Disconnect();
		Connect(this.Host, this.Port, this.VirtualHost, _user, _pwd);
		Thread.Sleep(100);
		isConnected = _session != null && (_session.Closed == false);
		i++;
		if (isConnected) {
			break;
		}
		OnMessageWorkerLogMessage(this.GetServiceName() +
			(isConnected ? " reconnected" :
			" still closed. Attempting " + (tryCnt - 1 - i) + " more times"));
	}
}

Visualizing economics

A nice set of graphics from Visualizing Economics, via Zero Hedge.

Articles on FIXatdl

With the formal introduction of FIXatdl at the last FPL’s EMEA meeting, v1.1 is now an official FIX standard. Here, then, are the latest articles published in FIXGlobal on the new standard.