DevWiki/websocket-sharp
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Modified Example1

Browse Source
This commit is contained in:
sta 2014-03-12 20:23:37 +09:00
parent 9a22fb4235
commit f068425b10
7 changed files with 292 additions and 289 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
Example1/AudioMessage.cs Normal file
View File

@ -0,0 +1,12 @@
using System;
namespace Example1
{
internal class AudioMessage
{
public uint user_id;
public byte ch_num;
public uint buffer_length;
public float [,] buffer_array;
}
}

419
Example1/AudioStreamer.cs
View File

@ -1,8 +1,5 @@
using Newtonsoft.Json; using Newtonsoft.Json;
using Newtonsoft.Json.Linq; using Newtonsoft.Json.Linq;
#if NOTIFY
using Notifications;
#endif
using System; using System;
using System.Collections; using System.Collections;
using System.Collections.Generic; using System.Collections.Generic;
@ -13,310 +10,194 @@ using WebSocketSharp;
namespace Example1 namespace Example1
{ {
public struct NfMessage internal class AudioStreamer : IDisposable
{
public string Summary;
public string Body;
public string Icon;
}
public class AudioMessage
{
public uint user_id;
public byte ch_num;
public uint buffer_length;
public float[,] buffer_array;
}
public class TextMessage
{
public uint? user_id;
public string name;
public string type;
public string message;
}
public class ThreadState
{
public bool Enabled { get; set; }
public AutoResetEvent Notification { get; private set; }
public ThreadState()
{
Enabled = true;
Notification = new AutoResetEvent(false);
}
}
public class AudioStreamer : IDisposable
{ {
private Dictionary<uint, Queue> _audioBox; private Dictionary<uint, Queue> _audioBox;
private Queue _msgQ;
private string _name;
private WaitCallback _notifyMsg;
private ThreadState _notifyMsgState;
private TimerCallback _sendHeartbeat;
private Timer _heartbeatTimer; private Timer _heartbeatTimer;
private uint? _user_id; private uint? _id;
private WebSocket _ws; private string _name;
private Notifier _notifier;
private WebSocket _websocket;
public AudioStreamer(string url) public AudioStreamer (string url)
{ {
_ws = new WebSocket(url); _websocket = new WebSocket (url);
_msgQ = Queue.Synchronized(new Queue());
_audioBox = new Dictionary<uint, Queue>();
_user_id = null;
configure(); _audioBox = new Dictionary<uint, Queue> ();
_heartbeatTimer = new Timer (sendHeartbeat, null, -1, -1);
_id = null;
_notifier = new Notifier ();
configure ();
} }
private void configure() private AudioMessage acceptBinaryMessage (byte [] value)
{ {
#if DEBUG var id = value.SubArray (0, 4).To<uint> (ByteOrder.Big);
_ws.Log.Level = LogLevel.Trace; var chNum = value.SubArray (4, 1) [0];
#endif var bufferLength = value.SubArray (5, 4).To<uint> (ByteOrder.Big);
_ws.OnOpen += (sender, e) => var bufferArray = new float [chNum, bufferLength];
{
var msg = createTextMessage("connection", String.Empty);
_ws.Send(msg);
};
_ws.OnMessage += (sender, e) => var offset = 9;
{ ((int) chNum).Times (
switch (e.Type) i => bufferLength.Times (
{ j => {
case Opcode.Text: bufferArray [i, j] = value.SubArray (offset, 4).To<float> (ByteOrder.Big);
var msg = parseTextMessage(e.Data); offset += 4;
_msgQ.Enqueue(msg); }));
break;
case Opcode.Binary:
var audioMsg = parseAudioMessage(e.RawData);
if (audioMsg.user_id == _user_id) goto default;
if (_audioBox.ContainsKey(audioMsg.user_id))
{
_audioBox[audioMsg.user_id].Enqueue(audioMsg.buffer_array);
}
else
{
var q = Queue.Synchronized(new Queue());
q.Enqueue(audioMsg.buffer_array);
_audioBox.Add(audioMsg.user_id, q);
}
break;
default:
break;
}
};
_ws.OnError += (sender, e) => return new AudioMessage {
{ user_id = id,
enNfMessage("[AudioStreamer] error", "WS: Error: " + e.Message, "notification-message-im"); ch_num = chNum,
}; buffer_length = bufferLength,
buffer_array = bufferArray
_ws.OnClose += (sender, e) =>
{
enNfMessage
(
"[AudioStreamer] disconnect",
String.Format("WS: Close({0}: {1})", e.Code, e.Reason),
"notification-message-im"
);
};
//_ws.Compression = CompressionMethod.Deflate;
_notifyMsgState = new ThreadState();
_notifyMsg = (state) =>
{
while (_notifyMsgState.Enabled || _msgQ.Count > 0)
{
Thread.Sleep(500);
if (_msgQ.Count > 0)
{
NfMessage msg = (NfMessage)_msgQ.Dequeue();
#if NOTIFY
Notification nf = new Notification(msg.Summary,
msg.Body,
msg.Icon);
nf.AddHint("append", "allowed");
nf.Show();
#else
Console.WriteLine("{0}: {1}", msg.Summary, msg.Body);
#endif
}
}
_notifyMsgState.Notification.Set();
};
_sendHeartbeat = (state) =>
{
var msg = createTextMessage("heartbeat", String.Empty);
_ws.Send(msg);
}; };
} }
private byte[] createAudioMessage(float[,] buffer_array) private NotificationMessage acceptTextMessage (string value)
{ {
List<byte> msg = new List<byte>(); var json = JObject.Parse (value);
var id = (uint) json ["user_id"];
uint user_id = (uint)_user_id; var name = (string) json ["name"];
int ch_num = buffer_array.GetLength(0); var type = (string) json ["type"];
int buffer_length = buffer_array.GetLength(1);
msg.AddRange(user_id.ToByteArray(ByteOrder.Big));
msg.Add((byte)ch_num);
msg.AddRange(((uint)buffer_length).ToByteArray(ByteOrder.Big));
ch_num.Times(i =>
{
buffer_length.Times(j =>
{
msg.AddRange(buffer_array[i, j].ToByteArray(ByteOrder.Big));
});
});
return msg.ToArray();
}
private string createTextMessage(string type, string message)
{
var msg = new TextMessage
{
user_id = _user_id,
name = _name,
type = type,
message = message
};
return JsonConvert.SerializeObject(msg);
}
private AudioMessage parseAudioMessage(byte[] data)
{
uint user_id = data.SubArray(0, 4).To<uint>(ByteOrder.Big);
byte ch_num = data.SubArray(4, 1)[0];
uint buffer_length = data.SubArray(5, 4).To<uint>(ByteOrder.Big);
float[,] buffer_array = new float[ch_num, buffer_length];
int offset = 9;
((int)ch_num).Times(i =>
{
buffer_length.Times(j =>
{
buffer_array[i, j] = data.SubArray(offset, 4).To<float>(ByteOrder.Big);
offset += 4;
});
});
return new AudioMessage
{
user_id = user_id,
ch_num = ch_num,
buffer_length = buffer_length,
buffer_array = buffer_array
};
}
private NfMessage parseTextMessage(string data)
{
JObject msg = JObject.Parse(data);
uint user_id = (uint)msg["user_id"];
string name = (string)msg["name"];
string type = (string)msg["type"];
string message; string message;
switch (type) if (type == "connection") {
{ var users = (JArray) json ["message"];
case "connection": var msg = new StringBuilder ("Now keeping connection:");
JArray users = (JArray)msg["message"]; foreach (JToken user in users)
StringBuilder sb = new StringBuilder("Now keeping connection\n"); msg.AppendFormat (
foreach (JToken user in users) "\n- user_id: {0} name: {1}", (uint) user ["user_id"], (string) user ["name"]);
{
sb.AppendFormat("user_id: {0} name: {1}\n", (uint)user["user_id"], (string)user["name"]);
}
message = sb.ToString().TrimEnd('\n');
break;
case "connected":
_user_id = user_id;
message = String.Format("user_id: {0} name: {1}", user_id, name);
break;
case "message":
message = String.Format("{0}: {1}", name, (string)msg["message"]);
break;
case "start_music":
message = String.Format("{0}: Started playing music!", name);
break;
default:
message = "Received unknown type message: " + type;
break;
}
return new NfMessage message = msg.ToString ();
{ }
Summary = String.Format("[AudioStreamer] {0}", type), else if (type == "connected") {
Body = message, _heartbeatTimer.Change (30000, 30000);
Icon = "notification-message-im" _id = id;
}; message = String.Format ("user_id: {0} name: {1}", id, name);
}
else if (type == "message")
message = String.Format ("{0}: {1}", name, (string) json ["message"]);
else if (type == "start_music")
message = String.Format ("{0}: Started playing music!", name);
else
message = "Received unknown type message.";
return new NotificationMessage {
Summary = String.Format ("AudioStreamer Message ({0})", type),
Body = message,
Icon = "notification-message-im"
};
} }
private void enNfMessage(string summary, string body, string icon) private void configure ()
{ {
var msg = new NfMessage #if DEBUG
{ _websocket.Log.Level = LogLevel.Trace;
Summary = summary, #endif
Body = body, _websocket.OnOpen += (sender, e) =>
Icon = icon _websocket.Send (createTextMessage ("connection", String.Empty));
_websocket.OnMessage += (sender, e) => {
if (e.Type == Opcode.Text)
_notifier.Notify (acceptTextMessage (e.Data));
else {
var msg = acceptBinaryMessage (e.RawData);
if (msg.user_id == _id)
return;
if (_audioBox.ContainsKey (msg.user_id)) {
_audioBox [msg.user_id].Enqueue (msg.buffer_array);
return;
}
var queue = Queue.Synchronized (new Queue ());
queue.Enqueue (msg.buffer_array);
_audioBox.Add (msg.user_id, queue);
}
}; };
_msgQ.Enqueue(msg); _websocket.OnError += (sender, e) =>
_notifier.Notify (
new NotificationMessage {
Summary = "AudioStreamer Error",
Body = e.Message,
Icon = "notification-message-im"
});
_websocket.OnClose += (sender, e) =>
_notifier.Notify (
new NotificationMessage {
Summary = String.Format ("AudioStreamer Disconnect ({0})", e.Code),
Body = e.Reason,
Icon = "notification-message-im"
});
} }
public void Connect() private byte [] createAudioMessage (float [,] bufferArray)
{ {
string name; var msg = new List<byte> ();
do
{ var id = (uint) _id;
Console.Write("Your name > "); var chNum = bufferArray.GetLength (0);
name = Console.ReadLine(); var bufferLength = bufferArray.GetLength (1);
msg.AddRange (id.ToByteArray (ByteOrder.Big));
msg.Add ((byte) chNum);
msg.AddRange (((uint) bufferLength).ToByteArray (ByteOrder.Big));
chNum.Times (
i => bufferLength.Times (
j => msg.AddRange (bufferArray [i, j].ToByteArray (ByteOrder.Big))));
return msg.ToArray ();
}
private string createTextMessage (string type, string message)
{
return JsonConvert.SerializeObject (
new TextMessage {
user_id = _id,
name = _name,
type = type,
message = message
});
}
private void sendHeartbeat (object state)
{
_websocket.Send (createTextMessage ("heartbeat", String.Empty));
}
public void Connect ()
{
do {
Console.Write ("Input your name> ");
_name = Console.ReadLine ();
} }
while (name == String.Empty); while (_name.Length == 0);
_name = name; _websocket.Connect ();
_ws.Connect();
ThreadPool.QueueUserWorkItem(_notifyMsg);
_heartbeatTimer = new Timer(_sendHeartbeat, null, 30 * 1000, 30 * 1000);
} }
public void Disconnect() public void Disconnect ()
{ {
var wait = new AutoResetEvent(false); var wait = new ManualResetEvent (false);
_heartbeatTimer.Dispose(wait); _heartbeatTimer.Dispose (wait);
wait.WaitOne(); wait.WaitOne ();
_ws.Close(); _websocket.Close ();
_notifier.Close ();
_notifyMsgState.Enabled = false;
_notifyMsgState.Notification.WaitOne();
} }
public void Dispose() public void Write (string message)
{ {
Disconnect(); _websocket.Send (createTextMessage ("message", message));
} }
public void Write(string data) void IDisposable.Dispose ()
{ {
var msg = createTextMessage("message", data); Disconnect ();
_ws.Send(msg);
}
public void Write(FileInfo file)
{
throw new NotImplementedException();
} }
} }
} }

8
Example1/Example1.csproj
View File

@ -34,7 +34,7 @@
<DebugType>full</DebugType> <DebugType>full</DebugType>
<Optimize>false</Optimize> <Optimize>false</Optimize>
<OutputPath>bin\Debug_Ubuntu</OutputPath> <OutputPath>bin\Debug_Ubuntu</OutputPath>
<DefineConstants>DEBUG;NOTIFY</DefineConstants> <DefineConstants>DEBUG;UBUNTU</DefineConstants>
<ErrorReport>prompt</ErrorReport> <ErrorReport>prompt</ErrorReport>
<WarningLevel>4</WarningLevel> <WarningLevel>4</WarningLevel>
<Externalconsole>true</Externalconsole> <Externalconsole>true</Externalconsole>
@ -46,7 +46,7 @@
<ErrorReport>prompt</ErrorReport> <ErrorReport>prompt</ErrorReport>
<WarningLevel>4</WarningLevel> <WarningLevel>4</WarningLevel>
<Externalconsole>true</Externalconsole> <Externalconsole>true</Externalconsole>
<DefineConstants>NOTIFY</DefineConstants> <DefineConstants>UBUNTU</DefineConstants>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
<Reference Include="System" /> <Reference Include="System" />
@ -62,6 +62,10 @@
<Compile Include="AssemblyInfo.cs" /> <Compile Include="AssemblyInfo.cs" />
<Compile Include="AudioStreamer.cs" /> <Compile Include="AudioStreamer.cs" />
<Compile Include="Program.cs" /> <Compile Include="Program.cs" />
<Compile Include="AudioMessage.cs" />
<Compile Include="TextMessage.cs" />
<Compile Include="NotificationMessage.cs" />
<Compile Include="Notifier.cs" />
</ItemGroup> </ItemGroup>
<Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" /> <Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
<ItemGroup> <ItemGroup>

24
Example1/NotificationMessage.cs Normal file
View File

@ -0,0 +1,24 @@
using System;
namespace Example1
{
internal class NotificationMessage
{
public string Body {
get; set;
}
public string Icon {
get; set;
}
public string Summary {
get; set;
}
public override string ToString ()
{
return String.Format ("{0}: {1}", Summary, Body);
}
}
}

77
Example1/Notifier.cs Normal file
View File

@ -0,0 +1,77 @@
#if UBUNTU
using Notifications;
#endif
using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading;
namespace Example1
{
internal class Notifier : IDisposable
{
private bool _enabled;
private Queue<NotificationMessage> _queue;
private ManualResetEvent _waitHandle;
public Notifier ()
{
_enabled = true;
_queue = new Queue<NotificationMessage> ();
_waitHandle = new ManualResetEvent (false);
ThreadPool.QueueUserWorkItem (
state => {
while (_enabled || Count > 0) {
Thread.Sleep (500);
if (Count > 0) {
var msg = dequeue ();
#if UBUNTU
var nf = new Notification (msg.Summary, msg.Body, msg.Icon);
nf.AddHint ("append", "allowed");
nf.Show ();
#else
Console.WriteLine (msg);
#endif
}
}
_waitHandle.Set ();
});
}
public int Count {
get {
lock (((ICollection) _queue).SyncRoot) {
return _queue.Count;
}
}
}
private NotificationMessage dequeue ()
{
lock (((ICollection) _queue).SyncRoot) {
return _queue.Dequeue ();
}
}
public void Close ()
{
_enabled = false;
_waitHandle.WaitOne ();
_waitHandle.Close ();
}
public void Notify (NotificationMessage message)
{
lock (((ICollection) _queue).SyncRoot) {
_queue.Enqueue (message);
}
}
void IDisposable.Dispose ()
{
Close ();
}
}
}

29
Example1/Program.cs
View File

@ -5,29 +5,22 @@ namespace Example1
{ {
public class Program public class Program
{ {
public static void Main(string[] args) public static void Main (string [] args)
{ {
//using (AudioStreamer streamer = new AudioStreamer("ws://localhost:3000/socket")) using (var streamer = new AudioStreamer ("ws://agektmr.node-ninja.com:3000/socket"))
using (AudioStreamer streamer = new AudioStreamer("ws://agektmr.node-ninja.com:3000/socket")) //using (var streamer = new AudioStreamer ("ws://localhost:3000/socket"))
{ {
streamer.Connect(); streamer.Connect ();
Thread.Sleep(500); Console.WriteLine ("\nType \"exit\" to exit.\n");
Console.WriteLine("\nType \"exit\" to exit.\n"); while (true) {
Thread.Sleep (500);
string data; Console.Write ("> ");
while (true) var msg = Console.ReadLine ();
{ if (msg == "exit")
Thread.Sleep(500);
Console.Write("> ");
data = Console.ReadLine();
if (data == "exit")
{
break; break;
}
streamer.Write(data); streamer.Write (msg);
} }
} }
} }

12
Example1/TextMessage.cs Normal file
View File

@ -0,0 +1,12 @@
using System;
namespace Example1
{
internal class TextMessage
{
public uint? user_id;
public string name;
public string type;
public string message;
}
}