public void Can_roundtrip_DispatchItem_with_complex_event()
 {
     var body = new XDoc("msg").Elem("foo", "bar");
     var channel = new XUri("channel://foo.com/bar");
     var resource = new XUri("http://foo.com/baz/0");
     var origin1 = new XUri("http://foo.com/baz/1");
     var origin2 = new XUri("http://foo.com/baz/2");
     var recipient1 = new DispatcherRecipient(new XUri("http://recipient1"));
     var recipient2 = new DispatcherRecipient(new XUri("http://recipient2"));
     var via1 = new XUri("http://via1");
     var via2 = new XUri("http://via2");
     var ev = new DispatcherEvent(body, channel, resource, origin1, origin2);
     ev = ev.WithRecipient(false, recipient1, recipient2).WithVia(via1).WithVia(via2);
     var item = new DispatchItem(
         new XUri("http://foo"),
         ev,
         "abc"
     );
     var serializer = new DispatchItemSerializer();
     var stream = serializer.ToStream(item);
     var item2 = serializer.FromStream(stream);
     Assert.AreEqual(item.Uri, item2.Uri, "uri mismatch");
     Assert.AreEqual(item.Location, item2.Location, "location mismatch");
     Assert.AreEqual(item.Event.Id, item2.Event.Id, "id mismatch");
     Assert.AreEqual(body.ToCompactString(), item2.Event.AsDocument().ToCompactString(), "body mismatch");
     Assert.AreEqual(channel, item2.Event.Channel, "channel mismatch");
     Assert.AreEqual(resource, item2.Event.Resource, "resource mismatch");
     Assert.AreEqual(origin1, item2.Event.Origins[0], "first origin mismatch");
     Assert.AreEqual(origin2, item2.Event.Origins[1], "second origin mismatch");
     Assert.AreEqual(recipient1.Uri, item2.Event.Recipients[0].Uri, "first recipient mismatch");
     Assert.AreEqual(recipient2.Uri, item2.Event.Recipients[1].Uri, "second recipient mismatch");
     Assert.AreEqual(via1, item2.Event.Via[0], "first via mismatch");
     Assert.AreEqual(via2, item2.Event.Via[1], "second via mismatch");
 }

 public void Can_roundtrip_AtPath_via_GetRelativePathTo_with_trailing_slash_on_base()
 {
     var relative = "b/c";
     var baseUri = new XUri("http://foo/a/");
     var combined = baseUri.AtPath(relative);
     Assert.AreEqual(relative, combined.GetRelativePathTo(baseUri));
 }

 public void Deregister(XUri uri)
 {
     lock(_registry) {
         if(!_registry.ContainsKey(uri)) {
             return;
         }
         _registry.Remove(uri);
         _map.Remove(uri);
     }
 }

 public void Exact_Match()
 {
     XUriMap<string> map = new XUriMap<string>();
     string key = "foo/bar";
     map[new XUri("test://foo/bar")] = key;
     string match;
     int similarity;
     XUri test = new XUri("test://foo/bar");
     map.TryGetValue(test, out match, out similarity);
     Assert.AreEqual(key, match);
     Assert.AreEqual(test.MaxSimilarity, similarity);
 }

 //--- Constructors ---
 public HttpTransport(IDreamEnvironment env, XUri uri, AuthenticationSchemes authenticationSheme)
 {
     if(env == null) {
         throw new ArgumentNullException("env");
     }
     if(uri == null) {
         throw new ArgumentNullException("uri");
     }
     _env = env;
     _uri = uri.WithoutCredentialsPathQueryFragment();
     _minSimilarity = _uri.MaxSimilarity;
     _sourceInternal = _uri + " (internal)";
     _sourceExternal = _uri.ToString();
     _authenticationSheme = authenticationSheme;
 }

 //--- Methods ---
 public int GetScoreWithNormalizedUri(XUri uri, out XUri normalized)
 {
     normalized = uri;
     switch(uri.Scheme.ToLowerInvariant()) {
     case "http":
     case "https":
         return 1;
     case "ext-http":
         normalized = normalized.WithScheme("http");
         return int.MaxValue;
     case "ext-https":
         normalized = normalized.WithScheme("https");
         return int.MaxValue;
     default:
         return 0;
     }
 }

 //--- Constructors ---
 public HttpTransport(IDreamEnvironment env, XUri uri, AuthenticationSchemes authenticationSheme, string dreamInParamAuthtoken)
 {
     if(env == null) {
         throw new ArgumentNullException("env");
     }
     if(uri == null) {
         throw new ArgumentNullException("uri");
     }
     _env = env;
     _uri = uri.WithoutCredentialsPathQueryFragment();
     _minSimilarity = _uri.MaxSimilarity;
     _sourceInternal = _uri + " (internal)";
     _sourceExternal = _uri.ToString();
     _authenticationSheme = authenticationSheme;
     _dreamInParamAuthtoken = dreamInParamAuthtoken;
     this.ServerSignature = "Dream-HTTPAPI/" + DreamUtil.DreamVersion;
 }

 public void Can_roundtrip_DispatchItem()
 {
     var msg = new XDoc("msg");
     var channel = new XUri("channel://foo.com/bar");
     var origin = new XUri("http://foo.com/baz");
     var ev = new DispatcherEvent(msg, channel, origin);
     var item = new DispatchItem(
         new XUri("http://foo"),
         ev,
         "abc"
     );
     var serializer = new DispatchItemSerializer();
     var stream = serializer.ToStream(item);
     var item2 = serializer.FromStream(stream);
     Assert.AreEqual(item.Uri, item2.Uri, "uri mismatch");
     Assert.AreEqual(item.Location, item2.Location, "location mismatch");
     Assert.AreEqual(item.Event.Id, item2.Event.Id, "id mismatch");
 }

 public void Add_via_to_event()
 {
     XDoc msg = new XDoc("msg");
     XUri via1 = new XUri("http://foo.com/route1");
     XUri via2 = new XUri("http://foo.com/route2");
     DispatcherEvent ev1 = new DispatcherEvent(msg, new XUri("channel://foo.com/bar"), new XUri("http://foo.com/baz"));
     DispatcherEvent ev2 = ev1.WithVia(via1);
     Assert.AreEqual(0, ev1.Via.Length);
     Assert.AreEqual(1, ev2.Via.Length);
     Assert.AreEqual(via1, ev2.Via[0]);
     DispatcherEvent ev3 = ev2.WithVia(via2);
     Assert.AreEqual(2, ev3.Via.Length);
     Assert.AreEqual(via1, ev3.Via[0]);
     Assert.AreEqual(via2, ev3.Via[1]);
     DreamMessage ev3msg = ev3.AsMessage();
     Assert.AreEqual(msg, ev3msg.ToDocument());
     Assert.AreEqual(ev1.Id, ev3.Id);
     Assert.AreEqual("channel://foo.com/bar", ev3msg.Headers.DreamEventChannel);
     Assert.AreEqual("http://foo.com/baz", ev3msg.Headers.DreamEventOrigin[0]);
     Assert.AreEqual(via1.ToString(), ev3msg.Headers.DreamEventVia[0]);
 }

        public void Adding_new_sub_will_push_combined_set_to_upstream_subscriber()
        {
            var downstream = CreatePubSubService().WithInternalKey().AtLocalHost;
            var upstreamUri = new XUri("http://upstream/");
            var upstreamMock = MockPlug.Register(upstreamUri);
            XDoc downstreamSet = null;
            upstreamMock.Expect().Verb("POST").RequestDocument(_ => {
                downstreamSet = _;
                return true;
            }).Response(DreamMessage.Ok());
            var subscribeToDownstream = new XDoc("subscription-set")
                .Elem("uri.owner", upstreamUri)
                .Start("subscription")
                    .Attr("id", "1")
                    .Elem("channel", "pubsub://*/*")
                    .Start("recipient").Elem("uri", upstreamUri).End()
                .End();

            // subscribe to set updates from downstream, expect most recent version to be pushed
            var downstreamResponse = downstream.At("subscribers").PostAsync(subscribeToDownstream).Wait();
            Assert.IsTrue(downstreamResponse.IsSuccessful, downstreamResponse.ToText());
            Assert.IsTrue(upstreamMock.WaitAndVerify(TimeSpan.FromSeconds(10)), upstreamMock.VerificationFailure);

            // someone else posts a subscription, expect a new push including that sub
            upstreamMock.Expect().Verb("POST").RequestDocument(_ => {
                downstreamSet = _;
                return true;
            }).Response(DreamMessage.Ok());
            var subscription = new XDoc("subscription-set")
                .Elem("uri.owner", "http://foo/")
                .Start("subscription")
                    .Attr("id", "1")
                    .Elem("channel", "foo://bar/baz")
                    .Start("recipient").Elem("uri", "http://foo/bar").End()
                .End();
            var subResponse = downstream.At("subscribers").PostAsync(subscription).Wait();
            Assert.IsTrue(subResponse.IsSuccessful, subResponse.ToText());
            Assert.IsTrue(upstreamMock.WaitAndVerify(TimeSpan.FromSeconds(10)), upstreamMock.VerificationFailure);
            Assert.AreEqual(1, downstreamSet["subscription[channel='foo://bar/baz']"].ListLength);
        }

        public Yield convert(DreamContext context, DreamMessage request, Result<DreamMessage> response)
        {
            string methodStr = context.GetParam("method", string.Empty);
            string aliasStr = context.GetParam("id", string.Empty);

            XUri secPage = new XUri( context.ServerUri ).At( "deki", "pages", "=Convert:" + methodStr );
            Plug _dekiSec = Plug.New( secPage );
            XDoc secDoc = _dekiSec.Get().ToDocument();

            string userURL = secDoc["user.author/@href"].AsInnerText;
            Plug _dekiUser = Plug.New( userURL );
            XDoc userDoc = _dekiUser.Get().ToDocument();
            string perms = userDoc["permissions.user"]["operations"].AsInnerText;
            if ( !perms.Contains("UNSAFECONTENT") ) {
                response.Return( DreamMessage.Forbidden("UNSAFECONTENT not enabled") );
                yield break;
            }

            XUri codePage = new XUri( context.ServerUri ).At( "deki","pages", "=Convert:" + methodStr,"contents");
            Plug _deki = Plug.New( codePage, new TimeSpan(0,0,15) );
            string bodyText = _deki.With("format", "xhtml").Get().ToDocument()["body"]["pre"].Contents;

            for ( int i = 0; i < toHTML.GetLength(0); i++ ) {
                bodyText = Regex.Replace( bodyText, toHTML[i,0], toHTML[i,1] );
            }
            //string codeStr = HttpUtility.HtmlDecode( bodyText );
            //XDoc doc = XDocFactory.From( codeStr, MimeType.ANY );
            //response.Return( DreamMessage.Ok( MimeType.TEXT, bodyText ) );

            Assembly scriptCode = CompileCode( bodyText, response );
            if ( scriptCode != null ) {
                XmlDocument outDoc = runScript( scriptCode, aliasStr );
                response.Return( DreamMessage.Ok( MimeType.XML, new XDoc(outDoc) ) );
            }
            yield break;
        }

 public void Can_mock_a_request_with_a_stream_body()
 {
     var tmp = Path.GetTempFileName();
     var payload = "blahblah";
     File.WriteAllText(tmp, payload);
     var message = DreamMessage.FromFile(tmp);
     var uri = new XUri("http://mock/post/stream");
     MockPlug.Setup(uri).Verb("POST")
         .WithMessage(m => m.ToText() == payload)
         .ExpectAtLeastOneCall();
     var response = Plug.New(uri).Post(message, new Result<DreamMessage>()).Wait();
     response.AssertSuccess();
     MockPlug.VerifyAll(1.Seconds());
 }

 public void Deregister_allows_reregister_of_uri()
 {
     XUri uri = new XUri("http://www.mindtouch.com/foo");
     int firstCalled = 0;
     MockPlug.Register(uri, delegate(Plug p, string v, XUri u, DreamMessage r, Result<DreamMessage> r2) {
         firstCalled++;
         r2.Return(DreamMessage.Ok());
     });
     Assert.IsTrue(Plug.New(uri).GetAsync().Wait().IsSuccessful);
     Assert.AreEqual(1, firstCalled);
     MockPlug.Deregister(uri);
     int secondCalled = 0;
     MockPlug.Register(uri, delegate(Plug p, string v, XUri u, DreamMessage r, Result<DreamMessage> r2) {
         secondCalled++;
         r2.Return(DreamMessage.Ok());
     });
     Assert.IsTrue(Plug.New(uri).GetAsync().Wait().IsSuccessful);
     Assert.AreEqual(1, firstCalled);
     Assert.AreEqual(1, secondCalled);
 }

 //--- Constructors ---
 internal MockPlugInvocation(string verb, XUri uri, DreamMessage request, DreamHeaders responseHeaders) {
     Verb = verb;
     Uri = uri;
     Request = request;
     ResponseHeaders = responseHeaders;
 }

 public DreamMessage SyncUri(XUri uri)
 {
     return Response(new XDoc("r").Elem("uri", uri.WithoutQuery()));
 }

 /// <summary>
 /// Create a new subscription set from a subscription set document.
 /// </summary>
 /// <param name="setDoc">Set Xml document.</param>
 /// <param name="location">location id.</param>
 /// <param name="accessKey">secret key for accessing the set.</param>
 public PubSubSubscriptionSet(XDoc setDoc, string location, string accessKey)
 {
     try {
         // Note: not using AsUri to avoid automatic local:// translation
         Owner = new XUri(setDoc["uri.owner"].AsText);
         MaxFailureDuration = (setDoc["@max-failure-duration"].AsDouble ?? 0).Seconds();
         var subscriptions = new List<PubSubSubscription>();
         foreach(XDoc sub in setDoc["subscription"]) {
             subscriptions.Add(new PubSubSubscription(sub, this));
         }
         Version = setDoc["@version"].AsLong;
         Subscriptions = subscriptions.ToArray();
         Location = location;
         AccessKey = accessKey;
         MaxFailures = UsesFailureDuration ? int.MaxValue : setDoc["@max-failures"].AsInt ?? MAX_FAILURES;
     } catch(Exception e) {
         throw new ArgumentException("Unable to parse subscription set: " + e.Message, e);
     }
 }

 public IEnumerator<IYield> Invoke(Plug plug, string verb, XUri uri, DreamMessage request, Result<DreamMessage> response)
 {
     var match = GetBestMatch(uri);
     yield return AsyncUtil.Fork(() => match.Invoke(plug, verb, uri,  MemorizeAndClone(request), response), new Result(TimeSpan.MaxValue));
 }

        public void PubSub_downstream_chaining_and_subscription_propagation()
        {
            XUri downstreamUri = new XUri("http://localhost/downstream");
            var mock = MockPlug.Register(downstreamUri);
            XDoc downstreamDoc = null;
            mock.Expect().Verb("GET").Uri(downstreamUri).Response(DreamMessage.Ok(new XDoc("subscription-set")
                .Elem("uri.owner", downstreamUri)
                .Start("subscription")
                   .Attr("id", "1")
                   .Elem("channel", "channel:///foo/*")
                   .Elem("uri.proxy", downstreamUri.At("publish"))
                   .Start("recipient").Elem("uri", "http:///foo/sub1").End()
                .End()));
            mock.Expect().Verb("POST").Uri(downstreamUri).RequestDocument(doc => {
                downstreamDoc = doc;
                return true;
            });

            // create service, which should subscribe to the upstream one
            Plug upstreamPubSub = CreatePubSubService(new XDoc("config").Start("downstream").Elem("uri", downstreamUri).End()).WithInternalKey().AtLocalHost;
            Assert.IsTrue(mock.WaitAndVerify(TimeSpan.FromSeconds(10)), mock.VerificationFailure);
            var subUri = downstreamDoc["subscription/recipient/uri"].AsUri.AsPublicUri();
            Plug.GlobalCookies.Update(DreamCookie.ParseAllSetCookieNodes(downstreamDoc["subscription/set-cookie"]), subUri);
            Assert.AreEqual("pubsub://*/*", downstreamDoc["subscription/channel"].AsText);
            Assert.AreEqual(upstreamPubSub.Uri.WithoutQuery(), downstreamDoc["uri.owner"].AsUri);
            Assert.AreEqual(upstreamPubSub.Uri.At("subscribers").WithoutQuery(), subUri.WithoutLastSegment());

            // check that our set is properly represented by the upstream service
            DreamMessage upstreamSet = upstreamPubSub.At("subscribers").GetAsync().Wait();
            Assert.IsTrue(upstreamSet.IsSuccessful);
            XDoc upstreamDoc = upstreamSet.ToDocument();
            Assert.AreEqual(1, upstreamDoc["subscription"].ListLength);
            Assert.AreEqual("channel:///foo/*", upstreamDoc["subscription/channel"].AsText);

            // update the subscription set on the upstream service
            XDoc set2 = new XDoc("subscription-set")
               .Elem("uri.owner", downstreamUri)
               .Start("subscription")
                   .Attr("id", "1")
                   .Elem("channel", "channel:///bar/*")
                   .Elem("uri.proxy", downstreamUri.At("publish"))
                   .Start("recipient").Elem("uri", "http:///foo/sub1").End()
               .End();
            DreamMessage subPush = Plug.New(subUri).PostAsync(set2).Wait();
            Assert.IsTrue(subPush.IsSuccessful);
            upstreamSet = upstreamPubSub.At("subscribers").GetAsync().Wait();
            Assert.IsTrue(upstreamSet.IsSuccessful);
            upstreamDoc = upstreamSet.ToDocument();
            Assert.AreEqual(1, upstreamDoc["subscription"].ListLength);
            Assert.AreEqual("channel:///bar/*", upstreamDoc["subscription/channel"].AsText);
        }

        private MockPlug.IMockInvokee GetBestMatch(XUri uri)
        {
            MockPlug.IMockInvokee invokee;

            // using _registry as our guard for both _map and _registry, since they are always modified in sync
            lock(_registry) {
                int result;
                _map.TryGetValue(uri, out invokee, out result);
            }
            if(invokee != null) {
                return invokee;
            }
            return uri.SchemeHostPort.EqualsInvariant(DEFAULT) ? DefaultInvokee : null;
        }

        public void Subscribing_to_pubsub_channel_results_in_combined_set_being_pushed()
        {
            var downstream = CreatePubSubService().WithInternalKey().AtLocalHost;
            var upstreamUri = new XUri("http://upstream/");
            var upstreamMock = MockPlug.Register(upstreamUri);
            XDoc downstreamSet = null;
            upstreamMock.Expect().Verb("POST").RequestDocument(_ => {
                downstreamSet = _;
                return true;
            }).Response(DreamMessage.Ok());
            var subscribeToDownstream = new XDoc("subscription-set")
                .Elem("uri.owner", upstreamUri)
                .Start("subscription")
                    .Attr("id", "1")
                    .Elem("channel", "foo://bar/baz/*")
                    .Start("recipient").Elem("uri", upstreamUri).End()
                .End()
                .Start("subscription")
                    .Attr("id", "1")
                    .Elem("channel", "pubsub://*/*")
                    .Start("recipient").Elem("uri", upstreamUri).End()
                .End();

            // subscribe to set updates from downstream, expect most recent version to be pushed
            var downstreamResponse = downstream.At("subscribers").PostAsync(subscribeToDownstream).Wait();
            var location = downstreamResponse.Headers.Location;
            Assert.IsTrue(downstreamResponse.IsSuccessful, downstreamResponse.ToText());
            Assert.IsTrue(upstreamMock.WaitAndVerify(TimeSpan.FromSeconds(10)), upstreamMock.VerificationFailure);
            Assert.AreEqual(1, downstreamSet["subscription[channel='foo://bar/baz/*']"].ListLength);

            subscribeToDownstream
                .Start("subscription")
                    .Attr("id", "3")
                    .Elem("channel", "foo://bar/bob")
                    .Start("recipient").Elem("uri", upstreamUri).End()
                .End();

            // change subscription, expect most recent version to be pushed
            upstreamMock.Expect().Verb("POST").RequestDocument(_ => {
                downstreamSet = _;
                return true;
            }).Response(DreamMessage.Ok());
            downstreamResponse = Plug.New(location).PutAsync(subscribeToDownstream).Wait();
            Assert.IsTrue(downstreamResponse.IsSuccessful, downstreamResponse.ToText());
            Assert.IsTrue(upstreamMock.WaitAndVerify(TimeSpan.FromSeconds(10)), upstreamMock.VerificationFailure);
            Assert.AreEqual(1, downstreamSet["subscription[channel='foo://bar/baz/*']"].ListLength);
            Assert.AreEqual(1, downstreamSet["subscription[channel='foo://bar/bob']"].ListLength);
        }