1.  
2. //  PeerConnection.cpp
3. //  AppSharing
4. //
5. //  Created by Faraz Khan on 4/17/14.
6. //
7. //
8.  
9. #include "SHPeerConnection.h"
10. #include <utility>
11. #include "talk/base/common.h"
12. #include "talk/base/json.h"
13. #include "talk/base/logging.h"
14. #include "talk/base/ssladapter.h"
15. #include "portable_time.h"
16. #include "SHPeerConnectionFactory.h"
17.  
18. namespace nexus {
19.     //
20.     //  test_client.cpp
21.     //  libjingle_examples
22.     //
23.     //  Created by Faraz Khan on 4/15/14.
24.     //
25.     //
26.    
27.     // Names used for a IceCandidate JSON object.
28.     const char kCandidateSdpMidName[] = "sdpMid";
29.     const char kCandidateSdpMlineIndexName[] = "sdpMLineIndex";
30.     const char kCandidateSdpName[] = "candidate";
31.  
32.  
33.     // Names used for a SessionDescription JSON object.
34.     const char kSessionDescriptionTypeName[] = "type";
35.     const char kSessionDescriptionSdpName[] = "sdp";
36.    
37.     class DummySetSessionDescriptionObserver
38.     : public webrtc::SetSessionDescriptionObserver {
39.     public:
40.         static DummySetSessionDescriptionObserver* Create(ILog * log,std::string sdp) {
41.             return
42.             new talk_base::RefCountedObject<DummySetSessionDescriptionObserver>(log,sdp);
43.         }
44.         virtual void OnSuccess() {
45.             logger_->OnLog("Successfully set SDP");
46.         }
47.         virtual void OnFailure(const std::string& error)
48.         {
49.             std::stringstream str;
50.             str << "Failure setting SDP : " << sdp_ << " with error " << error;
51.             logger_->OnLog(str.str());
52.            
53.         }
54.        
55.     protected:
56.         DummySetSessionDescriptionObserver(ILog * logger , std::string sdp) : logger_(logger), sdp_(sdp) { }
57.         ~DummySetSessionDescriptionObserver() {}
58.        
59.         ILog * logger_;
60.         std::string sdp_;
61.     };
62.    
63.     SHPeerConnection::SHPeerConnection(SHPeerConnectionMessageTransport * transport,
64.                                        SHPeerConnectionDataCallback * callback,
65.                                        SHPeerConnectionChannelType channel_id,std::string jid,
66.                                        ILog * logger)
67.     : channel_type_(channel_id), message_transport_(transport),
68.     data_callback_(callback),buddyjid_(jid),local_set_(false),
69.     is_receiver_(false),in_shutdown_(false),logger_(logger),
70.     ice_disconnected_time_ms_(PortableTickCount()),last_renegotiate_time_ms_(PortableTickCount()),
71.     last_process_time_(PortableTickCount()),
72.     did_have_active_connection_(false),started_ts_(PortableTickCount()),is_master_(false),
73.     callback_on_other_thread_(false),yet_another_thread_(NULL)
74.     {
75.         talk_base::LogMessage::LogTimestamps();
76.         talk_base::LogMessage::LogToDebug(talk_base::LS_ERROR);
77.        
78. #ifdef VERBOSE_PEERCONNECTION_DEBUG
79.         talk_base::LogMessage::LogToDebug(talk_base::LS_VERBOSE);
80.         streaminterface_ = new LogStreamInterface();
81.         streaminterface_->SignalLogReceived.connect(this, &SHPeerConnection::Log);
82.         talk_base::LogMessage::LogToStream(streaminterface_, talk_base::LS_INFO);
83. #endif
84.        
85.         thread_ = new talk_base::Thread();
86.         thread_->Release();
87.         thread_->Start();
88.        
89.         identity_service_ = new SHIdentityService(logger_,jid);
90.        
91.        
92.        
93.        
94.         //        server.uri = "stun:stun.l.google.com:19302";
95. //        server.uri = "stun:165.225.139.11:443";
96. //        servers.push_back(server);
97.        
98.         webrtc::PeerConnectionInterface::IceServer turnserver;
99.         webrtc::PeerConnectionInterface::IceServer backupstunserver;
100.  
101.         turnserver.uri = "turn:165.225.139.11:443";
102.         turnserver.username = "powwie";
103.         turnserver.password = "magic#mobile351";
104.         servers.push_back(turnserver);
105.        
106. #ifndef USE_ONLY_UDP_TURN
107.         turnserver.uri =  "turn:165.225.139.11:443?transport=tcp";
108.         servers.push_back(turnserver);
109.  
110.         backupstunserver.uri = "stun:stun.l.google.com:19302";
111.         servers.push_back(backupstunserver);
112. #endif
113.        
114. //        server.uri = "turn:165.225.139.11:443?transport=tcp";
115. //        server.username = "powwie";
116. //        server.password = "magic#mobile351";
117. //        servers.push_back(server);
118.  
119.  
120. //        talk_base::ThreadManager::Instance()->SetCurrentThread(thread_);
121.         process_thread_  = webrtc::ProcessThread::CreateProcessThread();
122.         process_thread_->RegisterModule(this);
123.         process_thread_->Start();
124.        
125.     }
126.    
127.     void SHPeerConnection::Log(std::string log)
128.     {
129.         if (logger_)
130.         {
131.            logger_->OnLog(log);
132.         }
133.     }
134.    
135.     SHPeerConnection::~SHPeerConnection() {
136.         Log("Peerconnection delete");
137.  
138.        
139.         if (!thread_->IsCurrent())
140.         {
141.             Log("Stopping since thread not current");
142.             thread_->Stop();
143.         }
144.        
145.         ASSERT(peer_connection_.get() == NULL);
146.     }
147.    
148.     bool SHPeerConnection::connection_active() const
149.     {
150.         return (peer_connection_.get() != NULL &&
151.                 (peer_connection_->ice_connection_state() == webrtc::PeerConnectionInterface::kIceConnectionConnected ||
152.                  peer_connection_->ice_connection_state() == webrtc::PeerConnectionInterface::kIceConnectionCompleted));
153.     }
154.    
155.    
156.     void SHPeerConnection::Reconnect()
157.     {
158.         thread_->Post(this,RENEGOTIATE);
159.     }
160.    
161.     void SHPeerConnection::Connect()
162.     {
163.        
164.         thread_->Post(this, SHPeerConnection::DO_CONNECT);
165.     }
166.        
167.         void SHPeerConnection::ReceivedMessageFromPeer(std::string message)
168.         {
169. //         this->OnMessageFromPeer(message);
170.        thread_->Post(this, RECV_MESSAGE_FROM_PEER, new talk_base::TypedMessageData<std::string>(message));
171.         }
172.    
173.     void SHPeerConnection::Shutdown()
174.     {
175.         Log("PeerConnection Shutdown ");
176.         thread_->Post(this,PEER_CONNECTION_CLOSED);
177.     }
178.    
179.     bool SHPeerConnection::InitializePeerConnection() {
180.         Log("Initializing Peer Connection");
181.         ASSERT(peer_connection_.get() == NULL);
182.        
183.        
184.        
185.         if (!SHPeerConnectionFactory::GetPeerConnectionFactory().get()) {
186.             //        main_wnd_->MessageBox("Error",
187.             //                              "Failed to initialize PeerConnectionFactory", true);
188.             DeletePeerConnection();
189.             return false;
190.         }
191.        
192. //        server.uri = "turn:165.225.139.11:443?transport=tcp";
193. //        servers.push_back(server);
194.  
195.         //        server.uri = "stun:stun.l.google.com:19302";
196.  
197.  
198.        
199.         webrtc::DataChannelInit dcinit;
200.         constraints.AddMandatory(webrtc::MediaConstraintsInterface::kOfferToReceiveAudio, webrtc::MediaConstraintsInterface::kValueFalse);
201.         constraints.AddMandatory(webrtc::MediaConstraintsInterface::kOfferToReceiveVideo, webrtc::MediaConstraintsInterface::kValueFalse);
202.         constraints.AddMandatory(webrtc::MediaConstraintsInterface::kIceRestart, webrtc::MediaConstraintsInterface::kValueTrue);
203.  
204.         if (channel_type_ == SHChannelTypeSCTP)
205.         {
206.             constraints.AddOptional(webrtc::MediaConstraintsInterface::kEnableDtlsSrtp, webrtc::MediaConstraintsInterface::kValueTrue);
207.  
208.             dcinit.reliable = true;
209.             dcinit.ordered = true;
210.             //dcinit.maxRetransmits = 0;
211.            
212.         }
213.         else
214.         {
215.             constraints.AddOptional(webrtc::MediaConstraintsInterface::kEnableRtpDataChannels,webrtc::MediaConstraintsInterface::kValueTrue);
216.             constraints.AddOptional(webrtc::MediaConstraintsInterface::kEnableDtlsSrtp, webrtc::MediaConstraintsInterface::kValueFalse);
217.             dcinit.reliable = false;
218.             dcinit.ordered = false;
219.         }
220.        
221.         peer_connection_ = SHPeerConnectionFactory::GetPeerConnectionFactory()->CreatePeerConnection(servers,
222.                                                                           &constraints,
223.                                                                           identity_service_,
224.                                                                           this);
225.        
226.         if (is_master_)
227.         {
228.             if (channel_type_ == SHChannelTypeSCTP)
229.                 datachannel_ = peer_connection_->CreateDataChannel("sctp_chan", &dcinit);
230.             else
231.                 datachannel_ = peer_connection_->CreateDataChannel("udp_chan", &dcinit);
232.            
233.            
234.             datachannel_->RegisterObserver(this);
235.         }
236.        
237.         if (!peer_connection_.get()) {
238.             DeletePeerConnection();
239.         }
240.         Log("Initialized Peer Connection");
241.         started_ts_ = PortableTickCount();
242.         return peer_connection_.get() != NULL;
243.     }
244.    
245.     void SHPeerConnection::DeletePeerConnection() {
246.         Log("Deleting peer connection");
247.         //Quit this thread first, as there maybe be inbound messages that havent been processed yet. Wait for actions
248.         //before you start killing this thread
249.        
250.         if (yet_another_thread_ != NULL)
251.         {
252.             if (yet_another_thread_->IsCurrent())
253.                 yet_another_thread_->Quit();
254.             else
255.                 yet_another_thread_->Stop();
256.            
257.             Log("Quit secondary inbound messaging thread");
258.         }
259.  
260.        
261.         in_shutdown_ = true;
262.         thread_->Quit();
263.         Log("Thread quit complete");
264.  
265.         bool neverstarted = (peer_connection_.get() == NULL);
266.        
267.        
268.  
269.        
270.         if (neverstarted && data_callback_)
271.             data_callback_->OnConnectionClosed(channel_type_);
272.        
273.         if (datachannel_.get())
274.         {
275.             datachannel_->UnregisterObserver();
276.         }
277.        
278. //        if (peer_connection_.get())
279. //            peer_connection_->Close();
280. //        
281.        
282.        
283.         process_thread_->DeRegisterModule(this);
284.         process_thread_->Stop();
285.         webrtc::ProcessThread::DestroyProcessThread(process_thread_);
286.        
287.         peer_connection_ = NULL;
288.         Log("Peerconnection = NULL DONE");
289.         data_callback_->OnShutdownComplete(channel_type_);
290.         ResetCallbacks();
291.     }
292.    
293.     void SHPeerConnection::OnError() {
294.         Log("PeerConnection error");
295.         thread_->Post(this, PEER_CONNECTION_ERROR);
296.     }
297.    
298.    
299.     void SHPeerConnection::OnMessage(const webrtc::DataBuffer &buffer)
300.     {
301. //        std::stringstream stream;
302. //        stream << "receving udp size " << buffer.size();
303. //        Log(stream.str());
304.         if (in_shutdown_)
305.             return;
306.        
307.         if (data_callback_)
308.         {
309.             if (callback_on_other_thread_)
310.                 yet_another_thread_->Post(this,ON_DATA_RECEIVED,new talk_base::TypedMessageData<std::string>(std::string(buffer.data.data(),buffer.data.length())));
311.             else
312.                 data_callback_->OnDataReceived(channel_type_, (uint8_t*)buffer.data.data(), (int)buffer.size());
313.         }
314.     }
315.    
316.     void SHPeerConnection::OnStateChange()
317.     {
318.                
319.                 if (!in_shutdown_)
320.                 {
321.                         std::stringstream stream;
322.                         stream << "Datachannel state changed : " << datachannel_->state();
323.                         Log(stream.str());
324.  
325.          
326.                         if (datachannel_->state() == webrtc::DataChannelInterface::kOpen && data_callback_)
327.                         {
328.                                 stream.clear();
329.                                 stream << "Datachannel opened: " << datachannel_->state();
330.                                 Log(stream.str());
331.                                 if (is_receiver_)
332.                     thread_->Post(this,ON_INCOMING_CONNECTION);
333.                                 else
334.                     thread_->Post(this,ON_CONNECTION_OPENED);
335.                         }
336.                         if (datachannel_->state() == webrtc::DataChannelInterface::kClosed && data_callback_)
337.                         {
338.                                 data_callback_->OnConnectionClosed(channel_type_);
339.                         }
340.                 }
341.        
342.     }
343.  
344.     void SHPeerConnection::OnIceChange()
345.     {
346.         Log("Ice Status Changed");
347.     }
348.    
349.     void SHPeerConnection::OnRenegotiationNeeded()
350.     {
351.         Log("Renegotiation Needed");
352.     }
353.    
354.     void SHPeerConnection::OnIceComplete()
355.     {
356.         Log("Ice State Complete");
357.     }
358.    
359.     int SHPeerConnection::AddRef()
360.     {
361.         return 0;
362.     }
363.    
364.     int SHPeerConnection::Release()
365.     {
366.         return 0;
367.     }
368.     void SHPeerConnection::OnIceConnectionChange(webrtc::PeerConnectionInterface::IceConnectionState new_state)
369.     {
370.         std::stringstream stream;
371.         stream << "Ice connection state : " << new_state;
372.         Log(stream.str());
373.        
374.         switch (new_state) {
375.  
376.             case webrtc::PeerConnectionInterface::kIceConnectionDisconnected:
377.                                 Log("Temporary disconnection, chillout");
378.                 ice_disconnected_time_ms_ = PortableTickCount();
379.                 last_renegotiate_time_ms_ = PortableTickCount();
380.                                 break;
381.                         case webrtc::PeerConnectionInterface::kIceConnectionFailed:
382.                         case webrtc::PeerConnectionInterface::kIceConnectionClosed:
383.                                 thread_->Post(this,ON_CONNECTION_CLOSED);
384.                                 break;
385.            
386.             case webrtc::PeerConnectionInterface::kIceConnectionCompleted:
387.             case webrtc::PeerConnectionInterface::kIceConnectionConnected:
388.                 Log("Ice Connection Established!");
389.                 did_have_active_connection_ = true;
390.                 break;
391.             default:
392.                 break;
393.         }
394.     }
395.    
396.     void SHPeerConnection::OnIceGatheringChange(webrtc::PeerConnectionInterface::IceGatheringState new_state)
397.     {
398. #ifdef VERBOSE_PEERCONNECTION_DEBUG        
399.         std::stringstream stream;
400.         stream << "Ice Gathering state : " << new_state;
401.         Log(stream.str());
402. #endif
403.  
404.     }
405.    
406.     void SHPeerConnection::OnStateChange(webrtc::PeerConnectionObserver::StateType state_changed)
407.     {
408.        
409. #ifdef VERBOSE_PEERCONNECTION_DEBUG
410.         std::stringstream stream;
411.         stream << "Peer connection state : " << state_changed;
412.         Log(stream.str());
413. #endif
414.        
415.     }
416.    
417.    
418.    
419.     void SHPeerConnection::OnDataChannel(webrtc::DataChannelInterface* data_channel)
420.     {
421.         datachannel_ = data_channel;
422.         data_channel->AddRef();
423.         datachannel_->RegisterObserver(this);
424.         Log("Incoming Data channel established!");
425.     }
426.    
427.    
428.    
429.     // Called when a remote stream is added
430.     void SHPeerConnection::OnAddStream(webrtc::MediaStreamInterface* stream) {
431.         LOG(LERROR) << __FUNCTION__ << " " << stream->label();
432.        
433.         stream->AddRef();
434.        
435.         thread_->Post(this, NEW_STREAM_ADDED, new talk_base::TypedMessageData<webrtc::MediaStreamInterface*>(stream));
436.     }
437.    
438.    
439.     void SHPeerConnection::OnRemoveStream(webrtc::MediaStreamInterface* stream) {
440.         LOG(LERROR) << __FUNCTION__ << " " << stream->label();
441.         stream->AddRef();
442.        
443.         thread_->Post(this, STREAM_REMOVED, new talk_base::TypedMessageData<webrtc::MediaStreamInterface*>(stream));
444.        
445.     }
446.    
447.     void SHPeerConnection::OnIceCandidate(const webrtc::IceCandidateInterface* candidate) {
448.        
449.         Json::StyledWriter writer;
450.         Json::Value jmessage;
451.        
452.         jmessage[kCandidateSdpMidName] = candidate->sdp_mid();
453.         jmessage[kCandidateSdpMlineIndexName] = candidate->sdp_mline_index();
454.         std::string sdp;
455.         if (!candidate->ToString(&sdp)) {
456.             Log("Failed to serialize candidate");
457.             return;
458.         }
459.         jmessage[kCandidateSdpName] = sdp;
460.         SendMessage(writer.write(jmessage));
461.     }
462.    
463.     void SHPeerConnection::ResetCallbacks()
464.     {
465.         data_callback_ = NULL;
466.         message_transport_ = NULL;
467.         logger_ = NULL;
468.     }
469.    
470.     void SHPeerConnection::OnMessageFromPeer(const std::string message) {
471.        
472.         if (in_shutdown_)
473.             return;
474.        
475.         if (!peer_connection_.get()) {
476.            
477.             if (!InitializePeerConnection()) {
478.                 Log("Failed to initialize our PeerConnection instance");
479.                 return;
480.             }
481.         }
482.        
483.         Json::Reader reader;
484.         Json::Value jmessage;
485.         if (!reader.parse(message, jmessage)) {
486.             std::stringstream stream;
487.             stream << "Received unknown message. " << message;
488.             Log(stream.str());
489.             return;
490.         }
491.  
492. #ifdef VERBOSE_PEERCONNECTION_DEBUG
493.         std::stringstream stream;
494.         stream << "Got Msg: " << message.c_str();
495.          Log(stream.str());
496. #endif
497.        
498.         std::string type;
499.         std::string json_object;
500.        
501.         GetStringFromJsonObject(jmessage, kSessionDescriptionTypeName, &type);
502.         if (!type.empty()) {
503.  
504.             std::string sdp;
505.                         if (!GetStringFromJsonObject(jmessage, kSessionDescriptionSdpName, &sdp)) {
506.                                 Log("Can't parse received session description message.");
507.                                 return;
508.                         }
509.                        
510.                         webrtc::SdpParseError * parseError = NULL;
511.                         webrtc::SessionDescriptionInterface* session_description(
512.                                                                                                                                          webrtc::CreateSessionDescription(type, sdp,parseError));
513.                         if (!session_description && !parseError) {
514.            
515.                 std::stringstream stream;
516.                 stream << "Error parsing message :" << parseError->description << " on line :"<<parseError->line;
517.                 Log(stream.str());
518.                 if (!did_have_active_connection_)
519.                                         thread_->Post(this,ON_CONNECTION_CLOSED);
520.                                
521.                                 return;
522.                         }
523.            
524.             std::stringstream stream;
525.             stream << "Received session description :" << message;
526.             Log(stream.str());
527.                        
528.            
529.                         peer_connection_->SetRemoteDescription(DummySetSessionDescriptionObserver::Create(logger_,sdp), session_description);
530.          
531.            
532.                         if (!is_master_&& session_description->type() == webrtc::SessionDescriptionInterface::kOffer) {
533.                                 is_receiver_ = true;
534.                                 peer_connection_->CreateAnswer(this, &constraints);
535.                         }
536.             return;
537.         } else  {
538.             local_set_ = true;
539.            if (local_set_)
540.                this->AddIceCandidate(message);
541.             else
542.                 pending_messages_.push_back(message);
543.         }
544.     }
545.    
546.     void SHPeerConnection::AddIceCandidate(std::string message)
547.     {
548.         Json::Reader reader;
549.         Json::Value jmessage;
550.         if (!reader.parse(message, jmessage)) {
551.             std::stringstream stream;
552.             stream << "Received unknown message. " << message;
553.             Log(stream.str());
554.             return;
555.         }
556.  
557.        
558.         std::string sdp_mid;
559.         int sdp_mlineindex = 0;
560.         std::string sdp;
561.         if (!GetStringFromJsonObject(jmessage, kCandidateSdpMidName, &sdp_mid) ||
562.             !GetIntFromJsonObject(jmessage, kCandidateSdpMlineIndexName,
563.                                   &sdp_mlineindex) ||
564.             !GetStringFromJsonObject(jmessage, kCandidateSdpName, &sdp)) {
565.             Log("Can't parse received message.");
566.             return;
567.         }
568.         talk_base::scoped_ptr<webrtc::IceCandidateInterface> candidate(
569.                                                                        webrtc::CreateIceCandidate(sdp_mid, sdp_mlineindex, sdp));
570.         if (!candidate.get()) {
571.             Log("Can't parse received candidate message.");
572.             return;
573.         }
574.         if (!peer_connection_->AddIceCandidate(candidate.get())) {
575.             Log("Failed to apply the received candidate");
576.             return;
577.         }
578.        
579.         std::stringstream stream;
580.         stream << " Received candidate :" << message;
581. //         Log(stream.str());
582.         return;
583.     }
584.    
585.     //        void SHPeerConnection::AddStreams() {
586.     //        if (active_streams_.find(kStreamLabel) != active_streams_.end())
587.     //            return;  // Already added.
588.     //
589.     //        talk_base::scoped_refptr<webrtc::AudioTrackInterface> audio_track(
590.     //                                                                          peer_connection_factory_->CreateAudioTrack(
591.     //                                                                                                                     kAudioLabel, peer_connection_factory_->CreateAudioSource(NULL)));
592.     //
593.     //    talk_base::scoped_refptr<webrtc::VideoTrackInterface> video_track(
594.     //                                                                      peer_connection_factory_->CreateVideoTrack(
595.     //                                                                                                                 kVideoLabel,
596.     //                                                                                                                 peer_connection_factory_->CreateVideoSource(OpenVideoCaptureDevice(),
597.     //                                                                                                                                                             NULL)));
598.     ////    main_wnd_->StartLocalRenderer(video_track);
599.    
600.     //        talk_base::scoped_refptr<webrtc::MediaStreamInterface> stream =
601.     //        peer_connection_factory_->CreateLocalMediaStream(kStreamLabel);
602.     //
603.     //
604.     //        stream->AddTrack(audio_track);
605.     //
606.     //    stream->AddTrack(video_track);
607.     //        if (!peer_connection_->AddStream(stream, NULL)) {
608.     //            LOG(LS_ERROR) << "Adding stream to PeerConnection failed";
609.     //        }
610.     //        typedef std::pair<std::string,
611.     //        talk_base::scoped_refptr<webrtc::MediaStreamInterface> >
612.     //        MediaStreamPair;
613.     //        active_streams_.insert(MediaStreamPair(stream->label(), stream));
614.     //    }
615.    
616.    
617.     int32_t SHPeerConnection::TimeUntilNextProcess()
618.     {
619.         return DEFAULT_PROCESS_TIME - (PortableTickCount() - last_process_time_);
620.     }
621.    
622.     int32_t SHPeerConnection::Process()
623.     {
624.         //test for disconnect timeout
625.         last_process_time_ = PortableTickCount();
626.         if (in_shutdown_)
627.             return 0;
628.        
629.                 if (did_have_active_connection_ && peer_connection_.get() != NULL)
630.                 {
631.                
632.                         bool is_disconnected =
633.             peer_connection_->ice_connection_state() == webrtc::PeerConnectionInterface::kIceConnectionDisconnected ||
634.                 peer_connection_->ice_connection_state() == webrtc::PeerConnectionInterface::kIceConnectionClosed ||
635.             peer_connection_->ice_connection_state() == webrtc::PeerConnectionInterface::kIceConnectionFailed ||
636.             peer_connection_->ice_connection_state() == webrtc::PeerConnectionInterface::kIceConnectionChecking;
637.        
638.                         if (time_delta_32(PortableTickCount(), ice_disconnected_time_ms_) > DEFAULT_DISCONNECT_TIMEOUT
639.                                 && is_disconnected)
640.                         {
641.                                 Log("Timeout happened, causing disconnect");
642.                                 thread_->Post(this,ON_CONNECTION_CLOSED);
643.                         }
644.             else if (time_delta_32(PortableTickCount(), last_renegotiate_time_ms_) > RENEGOTIATE_TIMEOUT
645.                      && is_disconnected)
646.             {
647.                 Log("Reconnecting....");
648.                 last_renegotiate_time_ms_ = PortableTickCount();
649.                this->Reconnect();
650.             }
651.         }
652.         else
653.         {
654.             if (time_delta_32(PortableTickCount(), started_ts_) > CONNECTION_ESTABLISHMENT_TIMEOUT)
655.             {
656.                 Log("Unable to establish conneciton, timeout");
657.                                 thread_->Post(this,ON_CONNECTION_CLOSED);
658.             }
659.         }
660.  
661.        
662.         return 0;
663.     }
664.    
665.  
666. bool SHPeerConnection::QueueDataToSend(uint8_t* data, int len)
667. {
668.     if (in_shutdown_)
669.         return false;
670.    
671.     std::string tosend((const char*)data,len);
672.     thread_->Post(this,SEND_DATA,new talk_base::TypedMessageData<std::string>(tosend));
673.     return true;
674. }
675.  
676.    
677.    bool SHPeerConnection::SendData(uint8_t * data, int len)
678.     {
679.         if (in_shutdown_)
680.             return false;
681.        
682.         if (datachannel_ && datachannel_.get() && datachannel_->state() == webrtc::DataChannelInterface::kOpen)
683.         {
684.            return datachannel_->Send(webrtc::DataBuffer(talk_base::Buffer(data,len),false));
685.         }
686.         return false;
687.     }
688.    
689. void SHPeerConnection::OnMessage(talk_base::Message *m)
690.     {
691.         if (in_shutdown_)
692.             return;
693.        
694.         switch (m->message_id) {
695.                         case RECV_MESSAGE_FROM_PEER:
696.             {
697.                 std::string msg;
698.                 if (m->pdata != NULL)
699.                     msg = ((talk_base::TypedMessageData<std::string>*)m->pdata)->data();
700.                
701.                 if (!msg.empty())
702.                 {
703.                     OnMessageFromPeer(msg);
704.                 }
705.                                
706.                                 break;
707.                         }
708.             case DO_CONNECT:
709.             {
710.                 is_master_ = true;
711.                 if (InitializePeerConnection())
712.                 {
713.                     peer_connection_->CreateOffer(this, &constraints);
714.                 }
715.                                 break;
716.             }
717.             case ON_DATA_RECEIVED:
718.             {
719.                 std::string msg;
720.                
721.                 if (m->pdata != NULL)
722.                     msg = ((talk_base::TypedMessageData<std::string>*)m->pdata)->data();
723.                
724.                 if (!msg.empty() && data_callback_)
725.                 {
726.                     data_callback_->OnDataReceived(channel_type_, (uint8_t*)msg.data(), (int)msg.length());;
727.                 }
728.                
729.                 break;
730.  
731.             }
732.        
733.             case SEND_DATA:
734.             {
735.                 std::string msg;
736.                
737.                 if (m->pdata != NULL)
738.                     msg = ((talk_base::TypedMessageData<std::string>*)m->pdata)->data();
739.                
740.                 if (!msg.empty())
741.                 {
742.                     webrtc::DataBuffer buf = webrtc::DataBuffer(msg);
743.                     datachannel_->Send(buf);
744.                 }
745.                
746.                 break;
747.             }
748.             case RENEGOTIATE:
749.             {
750.                 this->Renegotiate();
751.                 break;
752.             }
753.                         case ON_CONNECTION_CLOSED:
754.                         {
755.                                 if (data_callback_)
756.                     data_callback_->OnConnectionClosed(channel_type_);
757.              
758.                                 break;
759.             }
760.             case ON_INCOMING_CONNECTION:
761.             {
762.                 if (data_callback_)
763.                     data_callback_->OnIncomingConnection(channel_type_);
764.                 break;
765.             }
766.             case ON_CONNECTION_OPENED:
767.             {
768.                 if (data_callback_)
769.                     data_callback_->OnConnectionOpened(channel_type_);
770.                 break;
771.             }
772.                         case PEER_CONNECTION_CLOSED:
773.             {
774.                 Log("PEER_CONNECTION_CLOSED");
775.                                 DeletePeerConnection();
776.                 break;
777.             }
778.             case SEND_MESSAGE_TO_PEER: {
779.                
780.                 std::string msg;
781.                
782.                 if (m->pdata != NULL)
783.                     msg = ((talk_base::TypedMessageData<std::string>*)m->pdata)->data();
784.                
785.                 if (!msg.empty()) {
786. #ifdef VERBOSE_PEERCONNECTION_DEBUG
787.                     std::stringstream stream;
788.                     stream << "SEND_MESSAGE_TO_PEER: " << msg.c_str();
789.                      Log(stream.str());
790. #endif
791.                     if (message_transport_)
792.                         message_transport_->SendMessageToPeer(msg, buddyjid_, channel_type_);
793.                 }
794.                
795.                 break;
796.             }
797.                
798.             case PEER_CONNECTION_ERROR:
799.                 Log("Peer Conneciton Error");
800.                
801.                 break;
802.                
803.             case NEW_STREAM_ADDED: {
804.                 webrtc::MediaStreamInterface * stream =
805.                 ((talk_base::TypedMessageData<webrtc::MediaStreamInterface*>*)m->pdata)->data();
806.                 webrtc::VideoTrackVector tracks = stream->GetVideoTracks();
807.                 // Only render the first track.
808.                 if (!tracks.empty()) {
809.                     webrtc::VideoTrackInterface* track = tracks[0];
810.                     //                main_wnd_->StartRemoteRenderer(track);
811.                 }
812.                 stream->Release();
813.                 break;
814.             }
815.                
816.             case STREAM_REMOVED: {
817.                 // Remote peer stopped sending a stream.
818.                 webrtc::MediaStreamInterface * stream =
819.                 ((talk_base::TypedMessageData<webrtc::MediaStreamInterface*>*)m->pdata)->data();
820.                 stream->Release();
821.                 break;
822.             }
823.                
824.             default:
825.                 ASSERT(false);
826.                 break;
827.         }
828.     }
829.    
830.     void SHPeerConnection::OnSuccess(webrtc::SessionDescriptionInterface* desc)
831.     {
832.        
833.         std::string sdp;
834.         desc->ToString(&sdp);
835.                 Json::StyledWriter writer;
836.         Json::Value jmessage;
837.         jmessage[kSessionDescriptionTypeName] = desc->type();
838.                 webrtc::SdpParseError parserror;
839.                 webrtc::SessionDescriptionInterface * sdpinterface = webrtc::CreateSessionDescription(desc->type(), sdp, &parserror);
840.        
841.                 if (sdpinterface != NULL)
842.                 {
843.                         peer_connection_->SetLocalDescription(DummySetSessionDescriptionObserver::Create(logger_,sdp),
844.                                                                                                   sdpinterface);
845.                         jmessage[kSessionDescriptionSdpName] = sdp;
846.                         SendMessage(writer.write(jmessage));
847.                 }
848.                 else
849.                 {
850.                         std::stringstream str;
851.                         str << "Failed setting local description, cannot proceed! error " << parserror.description << " line: " << parserror.line;
852.                         logger_->OnLog(str.str());
853.                         if (!did_have_active_connection_)
854.                                 thread_->Post(this,ON_CONNECTION_CLOSED);
855.  
856.                 }
857.                
858.        
859.     }
860.    
861.     void SHPeerConnection::OnFailure(const std::string& error) {
862.         Log(error);
863.     }
864.    
865.     void SHPeerConnection::SendMessage(const std::string json_object) {
866. //        message_transport_->SendMessageToPeer(json_object, buddyjid_, channel_type_);
867.         thread_->Post(this, SEND_MESSAGE_TO_PEER, new talk_base::TypedMessageData<std::string>(json_object));
868.         //    main_wnd_->QueueUIThreadCallback(SEND_MESSAGE_TO_PEER, msg);
869.     }
870.    
871.     void SHPeerConnection::SetCallbackOnSecondaryThread(bool callback)
872.     {
873.         if (callback && !callback_on_other_thread_)
874.         {
875.             yet_another_thread_ = new talk_base::Thread();
876.             yet_another_thread_->Release();
877.             yet_another_thread_->Start();
878.         }
879.        
880.         callback_on_other_thread_ = callback;
881.     }
882.    
883.     void SHPeerConnection::Renegotiate()
884.     {
885.         if (peer_connection_.get() && is_master_)
886.         {
887.             peer_connection_->CreateOffer(this, &constraints);
888.         }
889.     }
890.    
891. }