1
我從我的項目顯著減少代碼，以便它的可複製並貼上但如果你想在一個控制檯項目調試它需要的NuGet包：安裝，包裝MsgPack.Cli。爲什麼名單強制所有新項目重複自己？
好吧，下面我已經在這是問題的評論線，所有我想知道的是爲什麼這個列表是_outgoingMessageQueue隊列中強迫重複。這是一種被捕獲的變量難題嗎？請給儘可能多的細節儘可能
using MsgPack.Serialization;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Collections;
namespace QueueTest
{
public class Message
{
public string Data { get; set; }
}
public class InternalFactoryMsg<T>
{
public T Data { get; set; }
public string Group { get; set; }
public List<byte[]> ReturnIds { get; set; }
}
public class FactoryHelpers
{
public static List<byte[]> GetReturnIdentities(List<byte[]> messageBytes, byte[] identity)
{
var response = new List<byte[]>();
foreach (byte[] part in messageBytes)
{
if (part != null && part.Length > 0)
response.Add(part);
else
break;
}
// may not need this with good routing, but can avoid errors
if (messageBytes.Count > 0 && messageBytes[0] == identity)
{
messageBytes.RemoveAt(0);
Console.WriteLine("[GetReturnIdentities]: Removed identity from start, check your routing!");
}
// no return identities, send empty list as these bytes will be the
// app message and command identity couple
if (response.Count == messageBytes.Count)
return new List<byte[]>();
return response;
}
public static byte[] SerializeData<T>(T appMsg)
{
var serializer = MessagePackSerializer.Get<T>();
using (var byteStream = new MemoryStream())
{
serializer.Pack(byteStream, appMsg);
return byteStream.ToArray();
}
}
public static T DeserializeData<T>(byte[] bytes)
{
try
{
var serializer = MessagePackSerializer.Get<T>();
using (var byteStream = new MemoryStream(bytes))
{
return serializer.Unpack(byteStream);
}
}
catch (Exception ex)
{
return default(T);
}
}
}
public class Factory: FactoryHelpers
{
protected ConcurrentQueue<KeyValuePair<string, List<byte[]>>> _outgoingMessageQueue { get; set; }
public ConcurrentQueue<KeyValuePair<string, List<byte[]>>> IncomingMessageQueue { get; set; }
public Factory()
{
_outgoingMessageQueue = new ConcurrentQueue<KeyValuePair<string, List<byte[]>>>();
IncomingMessageQueue = new ConcurrentQueue<KeyValuePair<string, List<byte[]>>>();
// add fake incoming message
var byteMsg = new List<byte[]>()
{
Encoding.Unicode.GetBytes("socket1"),
Encoding.Unicode.GetBytes(""),
Encoding.Unicode.GetBytes("data")
};
var msg = new KeyValuePair<string, List<byte[]>>("socket1", byteMsg);
IncomingMessageQueue.Enqueue(msg);
}
public void AddMessage<T>(InternalFactoryMsg<T> msg)
{
var msgBytes = msg.ReturnIds ?? new List<byte[]>();
msgBytes.Add(new byte[0]);
msgBytes.Add(Factory.SerializeData<T>(msg.Data));
_outgoingMessageQueue.Enqueue(new KeyValuePair<string, List<byte[]>>("socket2", msgBytes));
}
public List<KeyValuePair<string, List<byte[]>>> GetQueue()
{
return _outgoingMessageQueue.ToList();
}
public static T GetDataFromBytes<T>(List<byte[]> msgBytes)
{
// ignoring null checks etc
return DeserializeData<T>(msgBytes.Last());
}
}
public static class MessageLayer
{
public static Factory Factory = new Factory();
public static void Init()
{
Task.Factory.StartNew(u =>
{
while(true)
{
KeyValuePair<string, List<byte[]>> msg;
if(Factory.IncomingMessageQueue.TryDequeue(out msg))
{
var data = msg.Value.Last();
var returnIds = Factory.GetReturnIdentities(msg.Value, Encoding.Unicode.GetBytes(msg.Key));
IncomingCommands.HandleDataCommand(data, "test grp", returnIds);
}
// nice and slow for simulation
Thread.Sleep(400);
}
}, TaskCreationOptions.LongRunning);
}
public static void SendMessage(Message msg, string group, List<byte[]> returnIds)
{
var intMsg = new InternalFactoryMsg<Message>();
intMsg.Data = msg;
intMsg.Group = group;
intMsg.ReturnIds = returnIds;
Factory.AddMessage<Message>(intMsg);
}
}
public static class DataAccessor
{
public static List<Message> GetData(byte[] data)
{
return new List<Message>()
{
new Message() { Data = "magic" },
new Message() { Data = "data!" }
};
}
}
public static class IncomingCommands
{
public static void HandleDataCommand(byte[] data, string group, List<byte[]> returnIds)
{
List<Message> result;
// does big switch, gets data response
result = DataAccessor.GetData(data);
foreach (Message msg in result)
{
var local = msg;
var fix = new List<byte[]>(returnIds);
// THIS IS THE ISSUE
// comment out the following line and uncomment the one below to fix it
// but... why??? :O !!!
MessageLayer.SendMessage(local, group, returnIds);
//MessageLayer.SendMessage(local, group, fix);
}
// check the queue
Console.WriteLine("---------------------------");
Console.WriteLine("::Checking queue contents::");
var msgs = MessageLayer.Factory.GetQueue();
foreach(var m in msgs)
{
var check = Factory.GetDataFromBytes<Message>(m.Value);
Console.WriteLine("data -> " + check.Data);
}
}
}
public class Program
{
static void Main(string[] args)
{
MessageLayer.Init();
while(true)
{
Thread.Sleep(400);
}
}
}
}
如果你不能工作了，請了投票，因此備受關注。由於