9 files changed, 664 insertions, 85 deletions

diff --git a/jingle/notifier/base/sigslotrepeater.h b/jingle/notifier/base/sigslotrepeater.h
deleted file mode 100644
index cafb491..0000000
--- a/jingle/notifier/base/sigslotrepeater.h
+++ /dev/null
@@ -1,83 +0,0 @@
-// Copyright (c) 2009 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#ifndef JINGLE_NOTIFIER_BASE_SIGSLOTREPEATER_H_
-#define JINGLE_NOTIFIER_BASE_SIGSLOTREPEATER_H_
-
-// Repeaters are both signals and slots, which are designed as intermediate
-// pass-throughs for signals and slots which don't know about each other (for
-// modularity or encapsulation).  This eliminates the need to declare a signal
-// handler whose sole purpose is to fire another signal.  The repeater connects
-// to the originating signal using the 'repeat' method.  When the repeated
-// signal fires, the repeater will also fire.
-
-#include "talk/base/sigslot.h"
-
-namespace sigslot {
-
-template<class mt_policy = SIGSLOT_DEFAULT_MT_POLICY>
-class repeater0 : public signal0<mt_policy>,
-                  public has_slots<mt_policy> {
- public:
-  typedef signal0<mt_policy> base_type;
-  typedef repeater0<mt_policy> this_type;
-
-  repeater0() { }
-  explicit repeater0(const this_type& s) : base_type(s) { }
-
-  void reemit() { signal0<mt_policy>::emit(); }
-  void repeat(base_type &s) { s.connect(this, &this_type::reemit); }
-};
-
-template<class arg1_type, class mt_policy = SIGSLOT_DEFAULT_MT_POLICY>
-class repeater1 : public signal1<arg1_type, mt_policy>,
-                  public has_slots<mt_policy> {
- public:
-  typedef signal1<arg1_type, mt_policy> base_type;
-  typedef repeater1<arg1_type, mt_policy> this_type;
-
-  repeater1() { }
-  repeater1(const this_type& s) : base_type(s) { }
-
-  void reemit(arg1_type a1) { signal1<arg1_type, mt_policy>::emit(a1); }
-  void repeat(base_type& s) { s.connect(this, &this_type::reemit); }
-};
-
-template<class arg1_type, class arg2_type,
-         class mt_policy = SIGSLOT_DEFAULT_MT_POLICY>
-class repeater2 : public signal2<arg1_type, arg2_type, mt_policy>,
-                  public has_slots<mt_policy> {
- public:
-  typedef signal2<arg1_type, arg2_type, mt_policy> base_type;
-  typedef repeater2<arg1_type, arg2_type, mt_policy> this_type;
-
-  repeater2() { }
-  repeater2(const this_type& s) : base_type(s) { }
-
-  void reemit(arg1_type a1, arg2_type a2) {
-    signal2<arg1_type, arg2_type, mt_policy>::emit(a1, a2);
-  }
-  void repeat(base_type& s) { s.connect(this, &this_type::reemit); }
-};
-
-template<class arg1_type, class arg2_type, class arg3_type,
-         class mt_policy = SIGSLOT_DEFAULT_MT_POLICY>
-class repeater3 : public signal3<arg1_type, arg2_type, arg3_type, mt_policy>,
-                  public has_slots<mt_policy> {
- public:
-  typedef signal3<arg1_type, arg2_type, arg3_type, mt_policy> base_type;
-  typedef repeater3<arg1_type, arg2_type, arg3_type, mt_policy> this_type;
-
-  repeater3() { }
-  repeater3(const this_type& s) : base_type(s) { }
-
-  void reemit(arg1_type a1, arg2_type a2, arg3_type a3) {
-    signal3<arg1_type, arg2_type, arg3_type, mt_policy>::emit(a1, a2, a3);
-  }
-  void repeat(base_type& s) { s.connect(this, &this_type::reemit); }
-};
-
-}  // namespace sigslot
-
-#endif  // JINGLE_NOTIFIER_BASE_SIGSLOTREPEATER_H_
diff --git a/jingle/notifier/base/task_pump.cc b/jingle/notifier/base/task_pump.cc
index 52ffd26..9423ffa 100644
--- a/jingle/notifier/base/task_pump.cc
+++ b/jingle/notifier/base/task_pump.cc
@@ -12,12 +12,17 @@ TaskPump::TaskPump()
         ALLOW_THIS_IN_INITIALIZER_LIST(this)),
       posted_wake_(false) {}
 
-TaskPump::~TaskPump() {}
+TaskPump::~TaskPump() {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+}
 
 void TaskPump::WakeTasks() {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
   if (!posted_wake_) {
+    MessageLoop* current_message_loop = MessageLoop::current();
+    CHECK(current_message_loop);
     // Do the requested wake up.
-    MessageLoop::current()->PostTask(
+    current_message_loop->PostTask(
         FROM_HERE,
         scoped_runnable_method_factory_.NewRunnableMethod(
             &TaskPump::CheckAndRunTasks));
@@ -26,12 +31,14 @@ void TaskPump::WakeTasks() {
 }
 
 int64 TaskPump::CurrentTime() {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
   // Only timeout tasks rely on this function.  Since we're not using
   // libjingle tasks for timeout, it's safe to return 0 here.
   return 0;
 }
 
 void TaskPump::CheckAndRunTasks() {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
   posted_wake_ = false;
   // We shouldn't be using libjingle for timeout tasks, so we should
   // have no timeout tasks at all.
diff --git a/jingle/notifier/base/task_pump.h b/jingle/notifier/base/task_pump.h
index 806fed2..ff8411a 100644
--- a/jingle/notifier/base/task_pump.h
+++ b/jingle/notifier/base/task_pump.h
@@ -5,6 +5,7 @@
 #ifndef JINGLE_NOTIFIER_BASE_TASK_PUMP_H_
 #define JINGLE_NOTIFIER_BASE_TASK_PUMP_H_
 
+#include "base/non_thread_safe.h"
 #include "base/task.h"
 #include "talk/base/taskrunner.h"
 
@@ -23,6 +24,7 @@ class TaskPump : public talk_base::TaskRunner {
  private:
   void CheckAndRunTasks();
 
+  NonThreadSafe non_thread_safe_;
   ScopedRunnableMethodFactory<TaskPump> scoped_runnable_method_factory_;
   bool posted_wake_;
 
diff --git a/jingle/notifier/base/weak_xmpp_client.cc b/jingle/notifier/base/weak_xmpp_client.cc
new file mode 100644
index 0000000..4eedd9f
--- /dev/null
+++ b/jingle/notifier/base/weak_xmpp_client.cc
@@ -0,0 +1,41 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "jingle/notifier/base/weak_xmpp_client.h"
+
+#include "base/compiler_specific.h"
+
+namespace notifier {
+
+WeakXmppClient::WeakXmppClient(talk_base::TaskParent* parent)
+    : buzz::XmppClient(parent),
+      weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {}
+
+WeakXmppClient::~WeakXmppClient() {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  Invalidate();
+}
+
+void WeakXmppClient::Invalidate() {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  // We don't want XmppClient raising any signals once its invalidated.
+  SignalStateChange.disconnect_all();
+  SignalLogInput.disconnect_all();
+  SignalLogOutput.disconnect_all();
+  weak_ptr_factory_.InvalidateWeakPtrs();
+}
+
+base::WeakPtr<WeakXmppClient> WeakXmppClient::AsWeakPtr() {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  return weak_ptr_factory_.GetWeakPtr();
+}
+
+void WeakXmppClient::Stop() {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  // We don't want XmppClient used after it has been stopped.
+  Invalidate();
+  buzz::XmppClient::Stop();
+}
+
+}  // namespace notifier
diff --git a/jingle/notifier/base/weak_xmpp_client.h b/jingle/notifier/base/weak_xmpp_client.h
new file mode 100644
index 0000000..adda962
--- /dev/null
+++ b/jingle/notifier/base/weak_xmpp_client.h
@@ -0,0 +1,57 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+//
+// A thin wrapper around buzz::XmppClient that exposes weak pointers
+// so that users know when the buzz::XmppClient becomes invalid to use
+// (not necessarily only at destruction time).
+
+#ifndef JINGLE_NOTIFIER_BASE_WEAK_XMPP_CLIENT_H_
+#define JINGLE_NOTIFIER_BASE_WEAK_XMPP_CLIENT_H_
+#pragma once
+
+#include "base/basictypes.h"
+#include "base/non_thread_safe.h"
+#include "base/weak_ptr.h"
+#include "talk/xmpp/xmppclient.h"
+
+namespace talk_base {
+class TaskParent;
+}  // namespace
+
+namespace notifier {
+
+// buzz::XmppClient's destructor isn't marked virtual, but it inherits
+// from talk_base::Task, whose destructor *is* marked virtual, so we
+// can safely inherit from it.
+class WeakXmppClient : public buzz::XmppClient {
+ public:
+  explicit WeakXmppClient(talk_base::TaskParent* parent);
+
+  virtual ~WeakXmppClient();
+
+  // Returns a weak pointer that is invalidated when the XmppClient
+  // becomes invalid to use.
+  base::WeakPtr<WeakXmppClient> AsWeakPtr();
+
+  // Invalidates all weak pointers to this object.  (This method is
+  // necessary as calling Abort() does not always lead to Stop() being
+  // called, so it's not a reliable way to cause an invalidation.)
+  void Invalidate();
+
+ protected:
+  virtual void Stop();
+
+ private:
+  NonThreadSafe non_thread_safe_;
+  // We use our own WeakPtrFactory instead of inheriting from
+  // SupportsWeakPtr since we want to invalidate in other places
+  // besides the destructor.
+  base::WeakPtrFactory<WeakXmppClient> weak_ptr_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(WeakXmppClient);
+};
+
+}  // namespace notifier
+
+#endif  // JINGLE_NOTIFIER_BASE_WEAK_XMPP_CLIENT_H_
diff --git a/jingle/notifier/base/weak_xmpp_client_unittest.cc b/jingle/notifier/base/weak_xmpp_client_unittest.cc
new file mode 100644
index 0000000..705bb35
--- /dev/null
+++ b/jingle/notifier/base/weak_xmpp_client_unittest.cc
@@ -0,0 +1,124 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "jingle/notifier/base/weak_xmpp_client.h"
+
+#include "base/basictypes.h"
+#include "base/message_loop.h"
+#include "base/scoped_ptr.h"
+#include "base/weak_ptr.h"
+#include "jingle/notifier/base/task_pump.h"
+#include "talk/base/sigslot.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace notifier {
+
+namespace {
+
+class MockXmppDelegate : public sigslot::has_slots<> {
+ public:
+  virtual ~MockXmppDelegate() {}
+
+  MOCK_METHOD1(OnStateChange, void(buzz::XmppEngine::State));
+  MOCK_METHOD2(OnInputLog, void(const char*, int));
+  MOCK_METHOD2(OnOutputLog, void(const char*, int));
+};
+
+const buzz::XmppEngine::State kState = buzz::XmppEngine::STATE_OPEN;
+const char kInputLog[] = "input log";
+const char kOutputLog[] = "output log";
+
+class WeakXmppClientTest : public testing::Test {
+ protected:
+  WeakXmppClientTest() : task_pump_(new TaskPump()) {}
+
+  virtual ~WeakXmppClientTest() {}
+
+  void ConnectSignals(buzz::XmppClient* xmpp_client) {
+    xmpp_client->SignalStateChange.connect(
+        &mock_xmpp_delegate_, &MockXmppDelegate::OnStateChange);
+    xmpp_client->SignalLogInput.connect(
+        &mock_xmpp_delegate_, &MockXmppDelegate::OnInputLog);
+    xmpp_client->SignalLogOutput.connect(
+        &mock_xmpp_delegate_, &MockXmppDelegate::OnOutputLog);
+  }
+
+  void ExpectSignalCalls() {
+    EXPECT_CALL(mock_xmpp_delegate_, OnStateChange(kState));
+    EXPECT_CALL(mock_xmpp_delegate_,
+                OnInputLog(kInputLog, arraysize(kInputLog)));
+    EXPECT_CALL(mock_xmpp_delegate_,
+                OnOutputLog(kOutputLog, arraysize(kOutputLog)));
+  }
+
+  void RaiseSignals(buzz::XmppClient* xmpp_client) {
+    xmpp_client->SignalStateChange(kState);
+    xmpp_client->SignalLogInput(kInputLog, arraysize(kInputLog));
+    xmpp_client->SignalLogOutput(kOutputLog, arraysize(kOutputLog));
+  }
+
+  // Needed by TaskPump.
+  MessageLoop message_loop_;
+
+  scoped_ptr<TaskPump> task_pump_;
+  MockXmppDelegate mock_xmpp_delegate_;
+};
+
+TEST_F(WeakXmppClientTest, InvalidationViaInvalidate) {
+  ExpectSignalCalls();
+
+  WeakXmppClient* weak_xmpp_client = new WeakXmppClient(task_pump_.get());
+  ConnectSignals(weak_xmpp_client);
+
+  weak_xmpp_client->Start();
+  base::WeakPtr<WeakXmppClient> weak_ptr = weak_xmpp_client->AsWeakPtr();
+  EXPECT_TRUE(weak_ptr.get());
+  RaiseSignals(weak_ptr.get());
+
+  weak_xmpp_client->Invalidate();
+  EXPECT_FALSE(weak_ptr.get());
+  // We know that |weak_xmpp_client| is still valid at this point,
+  // although it should be entirely disconnected.
+  RaiseSignals(weak_xmpp_client);
+}
+
+TEST_F(WeakXmppClientTest, InvalidationViaStop) {
+  ExpectSignalCalls();
+
+  WeakXmppClient* weak_xmpp_client = new WeakXmppClient(task_pump_.get());
+  ConnectSignals(weak_xmpp_client);
+
+  weak_xmpp_client->Start();
+  base::WeakPtr<WeakXmppClient> weak_ptr = weak_xmpp_client->AsWeakPtr();
+  EXPECT_TRUE(weak_ptr.get());
+  RaiseSignals(weak_ptr.get());
+
+  weak_xmpp_client->Abort();
+  EXPECT_FALSE(weak_ptr.get());
+  // We know that |weak_xmpp_client| is still valid at this point,
+  // although it should be entirely disconnected.
+  RaiseSignals(weak_xmpp_client);
+}
+
+TEST_F(WeakXmppClientTest, InvalidationViaDestructor) {
+  ExpectSignalCalls();
+
+  WeakXmppClient* weak_xmpp_client = new WeakXmppClient(task_pump_.get());
+  ConnectSignals(weak_xmpp_client);
+
+  weak_xmpp_client->Start();
+  base::WeakPtr<WeakXmppClient> weak_ptr = weak_xmpp_client->AsWeakPtr();
+  EXPECT_TRUE(weak_ptr.get());
+  RaiseSignals(weak_ptr.get());
+
+  task_pump_.reset();
+  EXPECT_FALSE(weak_ptr.get());
+  // |weak_xmpp_client| is truly invalid at this point so we can't
+  // RaiseSignals() with it.
+}
+
+}  // namespace
+
+}  // namespace notifier
diff --git a/jingle/notifier/base/xmpp_connection.cc b/jingle/notifier/base/xmpp_connection.cc
new file mode 100644
index 0000000..c3597f9
--- /dev/null
+++ b/jingle/notifier/base/xmpp_connection.cc
@@ -0,0 +1,139 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "jingle/notifier/base/xmpp_connection.h"
+
+#include "base/compiler_specific.h"
+#include "base/logging.h"
+#include "base/message_loop.h"
+#include "base/string_piece.h"
+#include "jingle/notifier/base/chrome_async_socket.h"
+#include "jingle/notifier/base/task_pump.h"
+#include "jingle/notifier/base/weak_xmpp_client.h"
+#include "jingle/notifier/base/xmpp_client_socket_factory.h"
+#include "net/base/ssl_config_service.h"
+#include "talk/xmpp/xmppclientsettings.h"
+
+namespace notifier {
+
+namespace {
+
+buzz::AsyncSocket* CreateSocket(
+    const buzz::XmppClientSettings& xmpp_client_settings) {
+  bool use_fake_ssl_client_socket =
+      (xmpp_client_settings.protocol() == cricket::PROTO_SSLTCP);
+  net::ClientSocketFactory* const client_socket_factory =
+      new XmppClientSocketFactory(
+          net::ClientSocketFactory::GetDefaultFactory(),
+          use_fake_ssl_client_socket);
+  // The default SSLConfig is good enough for us for now.
+  const net::SSLConfig ssl_config;
+  // These numbers were taken from similar numbers in
+  // XmppSocketAdapter.
+  const size_t kReadBufSize = 64U * 1024U;
+  const size_t kWriteBufSize = 64U * 1024U;
+  // TODO(akalin): Use a real NetLog.
+  net::NetLog* const net_log = NULL;
+  return new ChromeAsyncSocket(
+      client_socket_factory, ssl_config,
+      kReadBufSize, kWriteBufSize, net_log);
+}
+
+}  // namespace
+
+XmppConnection::XmppConnection(
+    const buzz::XmppClientSettings& xmpp_client_settings,
+    Delegate* delegate, buzz::PreXmppAuth* pre_xmpp_auth)
+    : task_pump_(new TaskPump()),
+      on_connect_called_(false),
+      delegate_(delegate) {
+  DCHECK(delegate_);
+  // Owned by |task_pump_|, but is guaranteed to live at least as long
+  // as this function.
+  WeakXmppClient* weak_xmpp_client = new WeakXmppClient(task_pump_.get());
+  weak_xmpp_client->SignalStateChange.connect(
+      this, &XmppConnection::OnStateChange);
+  weak_xmpp_client->SignalLogInput.connect(
+      this, &XmppConnection::OnInputLog);
+  weak_xmpp_client->SignalLogOutput.connect(
+      this, &XmppConnection::OnOutputLog);
+  const char kLanguage[] = "en";
+  buzz::XmppReturnStatus connect_status =
+      weak_xmpp_client->Connect(xmpp_client_settings, kLanguage,
+                                CreateSocket(xmpp_client_settings),
+                                pre_xmpp_auth);
+  // buzz::XmppClient::Connect() should never fail.
+  DCHECK_EQ(connect_status, buzz::XMPP_RETURN_OK);
+  weak_xmpp_client->Start();
+  weak_xmpp_client_ = weak_xmpp_client->AsWeakPtr();
+}
+
+XmppConnection::~XmppConnection() {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  ClearClient();
+  MessageLoop* current_message_loop = MessageLoop::current();
+  CHECK(current_message_loop);
+  // We do this because XmppConnection may get destroyed as a result
+  // of a signal from XmppClient.  If we delete |task_pump_| here, bad
+  // things happen when the stack pops back up to the XmppClient's
+  // (which is deleted by |task_pump_|) function.
+  current_message_loop->DeleteSoon(FROM_HERE, task_pump_.release());
+}
+
+void XmppConnection::OnStateChange(buzz::XmppEngine::State state) {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  LOG(INFO) << "XmppClient state changed to " << state;
+  if (!weak_xmpp_client_.get()) {
+    LOG(DFATAL) << "weak_xmpp_client_ unexpectedly NULL";
+    return;
+  }
+  if (!delegate_) {
+    LOG(DFATAL) << "delegate_ unexpectedly NULL";
+    return;
+  }
+  switch (state) {
+    case buzz::XmppEngine::STATE_OPEN:
+      if (on_connect_called_) {
+        LOG(DFATAL) << "State changed to STATE_OPEN more than once";
+      } else {
+        delegate_->OnConnect(weak_xmpp_client_);
+        on_connect_called_ = true;
+      }
+      break;
+    case buzz::XmppEngine::STATE_CLOSED: {
+      int subcode = 0;
+      buzz::XmppEngine::Error error =
+          weak_xmpp_client_->GetError(&subcode);
+      const buzz::XmlElement* stream_error =
+          weak_xmpp_client_->GetStreamError();
+      ClearClient();
+      Delegate* delegate = delegate_;
+      delegate_ = NULL;
+      delegate->OnError(error, subcode, stream_error);
+      break;
+    }
+    default:
+      // Do nothing.
+      break;
+  }
+}
+
+void XmppConnection::OnInputLog(const char* data, int len) {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  LOG(INFO) << "XMPP Input: " << base::StringPiece(data, len);
+}
+
+void XmppConnection::OnOutputLog(const char* data, int len) {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  LOG(INFO) << "XMPP Output: " << base::StringPiece(data, len);
+}
+
+void XmppConnection::ClearClient() {
+  if (weak_xmpp_client_.get()) {
+    weak_xmpp_client_->Invalidate();
+    DCHECK(!weak_xmpp_client_.get());
+  }
+}
+
+}  // namespace notifier
diff --git a/jingle/notifier/base/xmpp_connection.h b/jingle/notifier/base/xmpp_connection.h
new file mode 100644
index 0000000..06d267b
--- /dev/null
+++ b/jingle/notifier/base/xmpp_connection.h
@@ -0,0 +1,97 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+//
+// A class that manages a connection to an XMPP server.
+
+#ifndef JINGLE_NOTIFIER_BASE_XMPP_CONNECTION_H_
+#define JINGLE_NOTIFIER_BASE_XMPP_CONNECTION_H_
+#pragma once
+
+#include "base/basictypes.h"
+#include "base/non_thread_safe.h"
+#include "base/scoped_ptr.h"
+#include "base/weak_ptr.h"
+#include "talk/base/sigslot.h"
+#include "talk/xmpp/xmppengine.h"
+#include "testing/gtest/include/gtest/gtest_prod.h"
+
+namespace buzz {
+class PreXmppAuth;
+class XmlElement;
+class XmppClientSettings;
+}  // namespace
+
+namespace talk_base {
+class Task;
+}  // namespace
+
+namespace notifier {
+
+class TaskPump;
+class WeakXmppClient;
+
+class XmppConnection : public sigslot::has_slots<> {
+ public:
+  class Delegate {
+   public:
+    virtual ~Delegate() {}
+
+    // Called (at most once) when a connection has been established.
+    // |base_task| can be used by the client as the parent of any Task
+    // it creates as long as it is valid (i.e., non-NULL).
+    virtual void OnConnect(base::WeakPtr<talk_base::Task> base_task) = 0;
+
+    // Called if an error has occurred (either before or after a call
+    // to OnConnect()).  No calls to the delegate will be made after
+    // this call.  Invalidates any weak pointers passed to the client
+    // by OnConnect().
+    //
+    // |error| is the code for the raised error.  |subcode| is an
+    // error-dependent subcode (0 if not applicable).  |stream_error|
+    // is non-NULL iff |error| == ERROR_STREAM.  |stream_error| is
+    // valid only for the lifetime of this function.
+    //
+    // Ideally, |error| would be set to something that is not
+    // ERROR_NONE, but due to inconsistent error-handling this doesn't
+    // always happen.
+    virtual void OnError(buzz::XmppEngine::Error error, int subcode,
+                         const buzz::XmlElement* stream_error) = 0;
+  };
+
+  // Does not take ownership of |delegate|, which may not be NULL.
+  // Takes ownership of |pre_xmpp_auth|, which may be NULL.
+  //
+  // TODO(akalin): Avoid the need for |pre_xmpp_auth|.
+  XmppConnection(const buzz::XmppClientSettings& xmpp_client_settings,
+                 Delegate* delegate, buzz::PreXmppAuth* pre_xmpp_auth);
+
+  // Invalidates any weak pointers passed to the delegate by
+  // OnConnect(), but does not trigger a call to the delegate's
+  // OnError() function.
+  ~XmppConnection();
+
+ private:
+  void OnStateChange(buzz::XmppEngine::State state);
+  void OnInputLog(const char* data, int len);
+  void OnOutputLog(const char* data, int len);
+
+  void ClearClient();
+
+  NonThreadSafe non_thread_safe_;
+  scoped_ptr<TaskPump> task_pump_;
+  base::WeakPtr<WeakXmppClient> weak_xmpp_client_;
+  bool on_connect_called_;
+  Delegate* delegate_;
+
+  FRIEND_TEST(XmppConnectionTest, RaisedError);
+  FRIEND_TEST(XmppConnectionTest, Connect);
+  FRIEND_TEST(XmppConnectionTest, MultipleConnect);
+  FRIEND_TEST(XmppConnectionTest, ConnectThenError);
+
+  DISALLOW_COPY_AND_ASSIGN(XmppConnection);
+};
+
+}  // namespace notifier
+
+#endif  // JINGLE_NOTIFIER_BASE_XMPP_CONNECTION_H_
diff --git a/jingle/notifier/base/xmpp_connection_unittest.cc b/jingle/notifier/base/xmpp_connection_unittest.cc
new file mode 100644
index 0000000..edf30e3
--- /dev/null
+++ b/jingle/notifier/base/xmpp_connection_unittest.cc
@@ -0,0 +1,195 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "jingle/notifier/base/xmpp_connection.h"
+
+#include <string>
+#include <vector>
+
+#include "base/basictypes.h"
+#include "base/message_loop.h"
+#include "base/weak_ptr.h"
+#include "jingle/notifier/base/weak_xmpp_client.h"
+#include "talk/xmpp/prexmppauth.h"
+#include "talk/xmpp/xmppclientsettings.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace buzz {
+class CaptchaChallenge;
+class Jid;
+}  // namespace buzz
+
+namespace talk_base {
+class CryptString;
+class SocketAddress;
+class Task;
+}  // namespace talk_base
+
+namespace notifier {
+
+using ::testing::_;
+using ::testing::Return;
+using ::testing::SaveArg;
+
+class MockPreXmppAuth : public buzz::PreXmppAuth {
+ public:
+  virtual ~MockPreXmppAuth() {}
+
+  MOCK_METHOD2(ChooseBestSaslMechanism,
+               std::string(const std::vector<std::string>&, bool));
+  MOCK_METHOD1(CreateSaslMechanism,
+               buzz::SaslMechanism*(const std::string&));
+  MOCK_METHOD4(StartPreXmppAuth,
+               void(const buzz::Jid&,
+                    const talk_base::SocketAddress&,
+                    const talk_base::CryptString&,
+                    const std::string&));
+  MOCK_CONST_METHOD0(IsAuthDone, bool());
+  MOCK_CONST_METHOD0(IsAuthorized, bool());
+  MOCK_CONST_METHOD0(HadError, bool());
+  MOCK_CONST_METHOD0(GetError, int());
+  MOCK_CONST_METHOD0(GetCaptchaChallenge, buzz::CaptchaChallenge());
+  MOCK_CONST_METHOD0(GetAuthCookie, std::string());
+};
+
+class MockXmppConnectionDelegate : public XmppConnection::Delegate {
+ public:
+  virtual ~MockXmppConnectionDelegate() {}
+
+  MOCK_METHOD1(OnConnect, void(base::WeakPtr<talk_base::Task>));
+  MOCK_METHOD3(OnError,
+               void(buzz::XmppEngine::Error, int, const buzz::XmlElement*));
+};
+
+class XmppConnectionTest : public testing::Test {
+ protected:
+  XmppConnectionTest() : mock_pre_xmpp_auth_(new MockPreXmppAuth()) {}
+
+  virtual ~XmppConnectionTest() {}
+
+  virtual void TearDown() {
+    // Clear out any messages posted by XmppConnections.
+    message_loop_.RunAllPending();
+  }
+
+  // Needed by XmppConnection.
+  MessageLoop message_loop_;
+  MockXmppConnectionDelegate mock_xmpp_connection_delegate_;
+  scoped_ptr<MockPreXmppAuth> mock_pre_xmpp_auth_;
+};
+
+TEST_F(XmppConnectionTest, CreateDestroy) {
+  XmppConnection xmpp_connection(buzz::XmppClientSettings(),
+                                 &mock_xmpp_connection_delegate_, NULL);
+}
+
+TEST_F(XmppConnectionTest, ImmediateFailure) {
+  // ChromeAsyncSocket::Connect() will always return false since we're
+  // not setting a valid host, but this gets bubbled up as ERROR_NONE
+  // due to XmppClient's inconsistent error-handling.
+  EXPECT_CALL(mock_xmpp_connection_delegate_,
+              OnError(buzz::XmppEngine::ERROR_NONE, 0, NULL));
+
+  XmppConnection xmpp_connection(buzz::XmppClientSettings(),
+                                 &mock_xmpp_connection_delegate_, NULL);
+}
+
+TEST_F(XmppConnectionTest, PreAuthFailure) {
+  EXPECT_CALL(*mock_pre_xmpp_auth_, StartPreXmppAuth(_, _, _, _));
+  EXPECT_CALL(*mock_pre_xmpp_auth_, IsAuthDone()).WillOnce(Return(true));
+  EXPECT_CALL(*mock_pre_xmpp_auth_, IsAuthorized()).WillOnce(Return(false));
+  EXPECT_CALL(*mock_pre_xmpp_auth_, HadError()).WillOnce(Return(true));
+  EXPECT_CALL(*mock_pre_xmpp_auth_, GetError()).WillOnce(Return(5));
+
+  EXPECT_CALL(mock_xmpp_connection_delegate_,
+              OnError(buzz::XmppEngine::ERROR_AUTH, 5, NULL));
+
+  XmppConnection xmpp_connection(
+      buzz::XmppClientSettings(), &mock_xmpp_connection_delegate_,
+      mock_pre_xmpp_auth_.release());
+}
+
+TEST_F(XmppConnectionTest, FailureAfterPreAuth) {
+  EXPECT_CALL(*mock_pre_xmpp_auth_, StartPreXmppAuth(_, _, _, _));
+  EXPECT_CALL(*mock_pre_xmpp_auth_, IsAuthDone()).WillOnce(Return(true));
+  EXPECT_CALL(*mock_pre_xmpp_auth_, IsAuthorized()).WillOnce(Return(true));
+  EXPECT_CALL(*mock_pre_xmpp_auth_, GetAuthCookie()).WillOnce(Return(""));
+
+  EXPECT_CALL(mock_xmpp_connection_delegate_,
+              OnError(buzz::XmppEngine::ERROR_NONE, 0, NULL));
+
+  XmppConnection xmpp_connection(
+      buzz::XmppClientSettings(), &mock_xmpp_connection_delegate_,
+      mock_pre_xmpp_auth_.release());
+}
+
+TEST_F(XmppConnectionTest, RaisedError) {
+  EXPECT_CALL(mock_xmpp_connection_delegate_,
+              OnError(buzz::XmppEngine::ERROR_NONE, 0, NULL));
+
+  XmppConnection xmpp_connection(buzz::XmppClientSettings(),
+                                 &mock_xmpp_connection_delegate_, NULL);
+
+  xmpp_connection.weak_xmpp_client_->
+      SignalStateChange(buzz::XmppEngine::STATE_CLOSED);
+}
+
+TEST_F(XmppConnectionTest, Connect) {
+  base::WeakPtr<talk_base::Task> weak_ptr;
+  EXPECT_CALL(mock_xmpp_connection_delegate_, OnConnect(_)).
+      WillOnce(SaveArg<0>(&weak_ptr));
+
+  {
+    XmppConnection xmpp_connection(buzz::XmppClientSettings(),
+                                   &mock_xmpp_connection_delegate_, NULL);
+
+    xmpp_connection.weak_xmpp_client_->
+        SignalStateChange(buzz::XmppEngine::STATE_OPEN);
+    EXPECT_EQ(xmpp_connection.weak_xmpp_client_.get(), weak_ptr.get());
+  }
+
+  EXPECT_EQ(NULL, weak_ptr.get());
+}
+
+TEST_F(XmppConnectionTest, MultipleConnect) {
+  EXPECT_DEBUG_DEATH({
+    base::WeakPtr<talk_base::Task> weak_ptr;
+    EXPECT_CALL(mock_xmpp_connection_delegate_, OnConnect(_)).
+        WillOnce(SaveArg<0>(&weak_ptr));
+
+    XmppConnection xmpp_connection(buzz::XmppClientSettings(),
+                                   &mock_xmpp_connection_delegate_, NULL);
+
+    xmpp_connection.weak_xmpp_client_->
+        SignalStateChange(buzz::XmppEngine::STATE_OPEN);
+    for (int i = 0; i < 3; ++i) {
+      xmpp_connection.weak_xmpp_client_->
+          SignalStateChange(buzz::XmppEngine::STATE_OPEN);
+    }
+
+    EXPECT_EQ(xmpp_connection.weak_xmpp_client_.get(), weak_ptr.get());
+  }, "more than once");
+}
+
+TEST_F(XmppConnectionTest, ConnectThenError) {
+  base::WeakPtr<talk_base::Task> weak_ptr;
+  EXPECT_CALL(mock_xmpp_connection_delegate_, OnConnect(_)).
+      WillOnce(SaveArg<0>(&weak_ptr));
+  EXPECT_CALL(mock_xmpp_connection_delegate_,
+              OnError(buzz::XmppEngine::ERROR_NONE, 0, NULL));
+
+  XmppConnection xmpp_connection(buzz::XmppClientSettings(),
+                                 &mock_xmpp_connection_delegate_, NULL);
+
+  xmpp_connection.weak_xmpp_client_->
+      SignalStateChange(buzz::XmppEngine::STATE_OPEN);
+  EXPECT_EQ(xmpp_connection.weak_xmpp_client_.get(), weak_ptr.get());
+
+  xmpp_connection.weak_xmpp_client_->
+      SignalStateChange(buzz::XmppEngine::STATE_CLOSED);
+  EXPECT_EQ(NULL, weak_ptr.get());
+}
+
+}  // namespace notifier