Open Mash Cross Reference
mash/tcl/degas/degasserver/degas-agent.tcl

   Class DegasAgent
   import RTP/Video
   import AddressBlock
   import DegasSession
   import Session/RTP/RTPGW
   import MediaAgent/RTPGW
   
   #------------------------------------------------------------------

   # Class DegasAgent

  #

  # purpose : This is the class that encapsulate the underlying 

  #           network session and provide an interface for querying

  #           underlying sessions, plus a set of callbacks for events

  #           such as joining and leaving sessions.  This is analogous

  #           to RTPGWAgent (for MeGa)

  #

  # members

  #

  # dali_interp_ 

  #    A C++ DaliInterp object that handles the execution of Dali program.

  #

  # out_session_

  #    A Session/RTP/RTPGW/Video object that encapsulates the output 

  #    multicast session.

  #

  # encoder_ 

  #    The encoder for the output video stream.

  #

  # decoders_(src)

  #    The decoder for the incoming video stream from $src.  Note that 

  #    since different sources can have different formats, we may have

  #    more than one decoders.  I need to find out how to share decoders

  #    between sources, if it is possible.

  #     

  # encoder_buffer_

  #    The buffers for encoder.  This get created with the encoder, and

  #    is not used.  We need this as a member so that we can free it 

  #    later.

  #

  # decoder_buffers_(src)

  #    The buffers for decoding.  This get created for each decoders.

  #

  # max_num_of_sources_

  #    Maximum number of sources we are suppose to serve.

  #

  # source_masks_

  #    The list of masks, a source will be serve iff it's name or ipaddr match

  #    one of the mask.  The syntax follows tcl "string match" syntax.

  #

  # num_of_sources_

  #    Number of sources currently being served.

  #

  # sources_

  #    The list of all active sources.

  #

  # rtp_

  #    RTP object that helps with various RTP functions, such as mapping

  #    source type to decoder.

  #------------------------------------------------------------------

  
  DegasAgent public init {aglp service program_spec out_spec} {
  
      # Creates input and output session

      
      $self instvar out_session_ in_session_ aglp_ service_
  
          set aglp_ $aglp
          set service_ $service
      # XXX : Need to fix this.

  
      set in_spec [$self get_program_value $program_spec "input_session"]
  
      set out_session_ [new DegasSession/Video $self $out_spec]
          set out_srcid [$out_session_ random-srcid [localaddr]]
          puts "Out session created : $out_session_ $out_spec"
          set out_src_mgr [new MediaAgent/RTPGW $self]
      $out_session_ sm $out_src_mgr
  #    [$out_session_ set netmgr_] set-subscription-level -1

          $out_src_mgr create-local $out_srcid [localaddr]
  
          $self instvar local_src_
          set local_src_ [$out_src_mgr local]
  
          global bps_
          set client_addr [$service get_client_addr]
          rate_variable bps_($client_addr) 0.1
          $self instvar start_nb_
          set start_nb_($client_addr) 0
          set bps_($client_addr) 0
          $self instvar report_rate_timer_
          set report_rate_timer_($local_src_) [new MashTimer "periodic" 5000 \
                  "$self report_rate $local_src_ $client_addr"]
  
      set in_session_ [new DegasSession/Video $self $in_spec]
          set in_srcid [$in_session_ random-srcid [localaddr]]
          puts "In session created : $in_session_ $in_spec"
          set in_src_mgr [new MediaAgent/RTPGW $self]
      $in_session_ sm $in_src_mgr
  #    [$in_session_ set netmgr_] set-subscription-level 0

         $in_src_mgr create-local $in_srcid [localaddr]
 
     #set out_session_ [new Session/RTP/RTPGW/Video $self]

     #set out_buffer_ [new BufferPool]

     #$out_session_ buffer-pool $out_buffer_

     #$out_session_ loopback-layer -1

     #set address_block [new AddressBlock $out_spec/1]

     #$out_session_ reset $address_block

     #delete $address_block

 
     #set in_session_ [new Session/RTP/RTPGW/Video $self]

     #set in_buffer_ [new BufferPool]

     #$in_session_ buffer-pool $in_buffer_

     #$in_session_ loopback-layer -1

         #set address_block [new AddressBlock $in_spec/1]

         #$in_session_ reset $address_block

         #delete $address_block

 
     
     # Setup the pipline dali_interp->encoder->session

 
     $self instvar dali_interp_
     $self instvar encoder_
     $self instvar encoder_buffer_
 
     set dali_interp_ [new DaliInterp]
 
     set res_mgr [DegasResourceManager info instances]
     $dali_interp_ set_resource_manager $res_mgr
 
     $self log_cpu_usage $res_mgr 
 
     $self init_interp $program_spec
     set outfmt [$self get_program_value $program_spec "output_format"]
     set encoder_ [new Module/VideoEncoder/UncompressedTo${outfmt}]
     if {$encoder_ == ""} {
         puts stderr "unable to create encoder for format $outfmt"
         puts stderr "creating default encoder for H261"
         set encoder_ [new Module/VideoEncoder/UncompressedToH261]
     } else {
                 puts stderr "encoder for $outfmt created"
     }
     $dali_interp_ target $encoder_
     $encoder_ target [$out_session_ get dih]
     set encoder_buffer_ [new BufferPool/RTP]
     $encoder_ buffer-pool $encoder_buffer_
     
     $encoder_buffer_ srcid $out_srcid
 
     $self instvar max_num_of_sources_
     $self instvar source_masks_
     $self instvar num_of_sources_
     $self instvar sources_
     $self instvar rtp_
 
     set max_num_of_sources_ [$self get_program_value $program_spec \
         "max_num_of_sources"]
     set source_masks_ [$self get_program_value $program_spec \
         "sources"]
     set num_of_sources_ 0
     set sources_ {}
     set rtp_ [new RTP/Video]
 
         puts "max number of sources $max_num_of_sources_"
 }
 
 
 #------------------------------------------------------------------

 # DegasAgent::destroy {}

 #

 # purpose : This is the destructor for DegasAgent.  Agent is deleted

 #           when there are no more client listening.

 # input   : none

 # output  : none

 #------------------------------------------------------------------

 
 DegasAgent public destroy {} {
 
         $self instvar sources_
         foreach src $sources_ {
                 $self deactivate $src
         }
 
     $self instvar rtp_
     delete $rtp_
 
     $self instvar out_session_
     delete [$out_session_ sm]
     delete $out_session_
 
     #$self instvar out_buffer_

     #delete $out_buffer_

 
     $self instvar in_session_
     delete [$in_session_ sm]
     delete $in_session_
 
     #$self instvar in_buffer_

     #delete $in_buffer_

 
     $self instvar dali_interp_
     delete $dali_interp_
 
     $self instvar encoder_
     delete $encoder_
 
     $self instvar encoder_buffer_
     delete $encoder_buffer_
 
         $self instvar report_rate_timer_ local_src_ service_
         global bps_ 
         unset bps_([$service_ get_client_addr])
         delete $report_rate_timer_($local_src_)
 }
 
 #------------------------------------------------------------------

 # DegasAgent::activate {src}

 #

 # purpose : This callback get called everytime a new source joins

 #           the session.

 # input   : src - Source object that just joined the session.

 # output  : none

 #------------------------------------------------------------------

 
 DegasAgent public activate {src} {
     $self instvar decoders_ decoder_buffers_ sources_
     $self instvar num_of_sources_ dali_interp_ rtp_ service_
 
         $service_ add_source [$src addr]
 
     if {![$self accept_source $src]} {
                 $self log "does not accept source $src"
         $src data-handler [new Module/VideoDecoder/Null]
         return
     }
 
     set ifmt [$rtp_ rtp_type [$src format]]
     set decoders_($src) [$self create_decoder $ifmt]
     if {$decoders_($src) == ""} {
         return
     }
     set decoder_buffers_($src) [new VidRep/Uncompressed]
     $decoders_($src) set_frame_buffer $decoder_buffers_($src)
     $decoder_buffers_($src) set_source_id $decoders_($src)
     
     lappend $sources_ $src
     incr num_of_sources_
 
     $decoders_($src) set src_ $src
     $src data-handler $decoders_($src)
     $decoders_($src) target $dali_interp_
     $dali_interp_ add_source $decoders_($src)
 }
 
 
 DegasAgent public report_rate {src addr} {
         global bps_ 
         $self instvar aglp_ 
 
         set bps_($addr) [$src layer-stat nb_]
         #puts "REPORT RATE $addr $bps_($addr)"

         $aglp_ set_bandwidth $addr $bps_($addr)
 }
 
 
 #------------------------------------------------------------------

 # DegasAgent::deactivate {src}

 #

 # purpose : This is called when a src leaves the session.

 # input   : src - the Source object that leaves the session.

 # output  : none.

 #------------------------------------------------------------------

 
 DegasAgent public deactivate {src} {
     $self instvar dali_interp_ 
     $self instvar sources_
     $self instvar decoders_
         $self instvar service_
 
         puts "Deactivating source $src [$src addr]"
         $service_ del_source [$src addr]
 
     # Remove the source

     set src_pos [lsearch -exact $sources_ $src]
     if {$src_pos == -1} {
        puts stderr "trying to delete a non-existance source $src"
        return
     }
     set sources_ [lreplace $sources_ $src_pos $src_pos]
 
     $dali_interp_ delete_source $decoders_($src)
     unset decoders_($src)
 }
 
 
 #------------------------------------------------------------------

 # DegasAgent::init_interp 

 #

 # purpose : initialize the interpreter using the program_spec.

 # input   : program_spec : this is a list of strings, representing

 #           the program to execute.

 # output  : none

 #------------------------------------------------------------------

 
 DegasAgent public init_interp {program_spec} {
     $self instvar dali_interp_
     foreach {key value} $program_spec {
         switch -glob $key {
             init_callback {
                 $dali_interp_ set_callback init $value
             }
 
             new_source_callback {
                 $dali_interp_ set_callback new_source $value
             }
 
             del_source_callback {
                 $dali_interp_ set_callback del_source $value
             }
 
             mouse_click_callback {
                 $dali_interp_ set_callback mouse_click $value
             }
 
             recv_frame_callback {
                 $dali_interp_ set_callback recv_frame $value
             }
 
             destroy_callback {
                 $dali_interp_ set_callback destroy $value
             }
 
             input_resize_callback {
                 $dali_interp_ set_callback input_resize $value
             }
 
             output_resize_callback {
                 $dali_interp_ set_callback output_resize $value
             }
 
             output_grayed_callback {
                 $dali_interp_ set_callback output_grayed $value
             }
 
             input_frame {
                 $dali_interp_ set_input_frame  [lindex $value 0]
                 $dali_interp_ set_input_width  [lindex $value 1]
                 $dali_interp_ set_input_height [lindex $value 2]
             }
 
             output_format {
                 $dali_interp_ set_output_format  $value
             }
 
             output_fps {
                 $dali_interp_ set_output_fps  $value
             }
 
             output_size {
                 $dali_interp_ set_output_size  $value
             }
 
             output_frame {
                 $dali_interp_ set_output_frame  [lindex $value 0]
                 $dali_interp_ set_output_width  [lindex $value 1]
                 $dali_interp_ set_output_height [lindex $value 2]
             }
 
             packages {
                 foreach package $value {
                     $dali_interp_ load_package $package
                 }
             }
 
             description {
                 $dali_interp_ set_description $value
             }
 
             source_id {
                 $dali_interp_ set_source_id $value
             }
 
             mouse_coord {
                 $dali_interp_ set_mouse_x_coord [lindex $value 0]
                 $dali_interp_ set_mouse_y_coord [lindex $value 1]
             }
 
                         resize_output_after {
                                         after [lindex $value 0] "$dali_interp_ resize_output_frame"
                         }
         }
     }
 }
 
 
 #------------------------------------------------------------------

 # DegasAgent::set_maxchannel

 # DegasAgent::unregister

 #

 # purpose : This is defined because Session/RTP/RTPGW calls them.

 #           I am not sure if we need these callbacks.

 # input   : don't care

 # output  : none

 #------------------------------------------------------------------

 
 DegasAgent public set_maxchannel {n} {
 
 }
 
 DegasAgent public unregister {src} {
 
 }
 
 DegasAgent public check_for_sources {} {
 
 }
 
 #------------------------------------------------------------------

 # DegasAgent::register

 #

 # purpose : a new source is detected.  creates a new control handler

 #           for this source.

 # input   : src

 # output  : none

 #------------------------------------------------------------------

 
 DegasAgent public register {src} {
 
     $self instvar out_session_
 
     set ctrl_handler [new Replicator/Packet/Copy]
     $src ctrl-handler $ctrl_handler
     $ctrl_handler add-target [$out_session_ get cih] 
 }
 
 
 
 #------------------------------------------------------------------

 # DegasAgent::get_program_value

 #

 # purpose : Given the program specification as a list, find the list

 #           element that corresponds to the key and return the value.

 # input   : program_spec - the program as a list.

 #           key - a string corresponds to the key we are interested 

 #                 in.  (e.g. "sources", "recv_frame_callback" ..)

 # output  : the value corresponds to the key

 #------------------------------------------------------------------

 
 DegasAgent private get_program_value { program_spec key } {
     set index [lsearch -glob $program_spec "$key"]
     if  {$index == -1} {
                 set value ""
     } else {
         set value [lindex $program_spec [expr $index + 1]]
     }
         return $value
 }
 
 
 #------------------------------------------------------------------

 # DegasAgent::create_decoder

 #

 # purpose : Creates a new decoder for the specified format

 # input   : format - the format for which we needs to creates decoder

 #           This is a string defined in tcl/net/rtp.tcl.  Only H261

 #           and JPEG is supported right now.

 # output  : the decoder

 #------------------------------------------------------------------

 
 DegasAgent private create_decoder { format } {
     switch $format {
         jpeg {
             return [new Module/VideoDecoder/JPEGToUncompressed]
         }
         h261 {
             return [new Module/VideoDecoder/H261ToUncompressed]
         }
         default {
             puts stderr "unknown input format $format. cannot create decoder."
             return ""
         }
     }
 }
 
 
 #------------------------------------------------------------------

 # DegasAgent::accept_source {src} 

 #

 # purpose : Check if we should accept this source, there are three

 #           conditions we should check : 

 #           - duplicate source ?

 #           - have we exceed the maximum number of source ?

 #           - does this src match the addresses given in mask ?

 # input   : src - the source to check.

 # output  : 1 if source is accepted, 0 otherwise.

 #------------------------------------------------------------------

 
 DegasAgent private accept_source {src} {
     $self instvar num_of_sources_
     $self instvar max_num_of_sources_
     $self instvar sources_
     $self instvar source_masks_
 
     if {$max_num_of_sources_ != "*" && 
         [regexp {[0-9]+} $max_num_of_sources_]} {
         if {$num_of_sources_ >= $max_num_of_sources_} {
             return 0
         }
     }
 
     if {[lsearch -exact $sources_ $src] != -1} {
         # duplicate source

         return 0
     }
 
     set src_ip_addr [$src addr]
     foreach mask $source_masks_ {
         # check ip address

         if {[string match $mask $src_ip_addr]} {
             return 1
         }
         # XXX : Need to figure out the hostname.

     }
     return 0
 }
 
 
 #------------------------------------------------------------------

 # DegasAgent::mouse_click_callback

 #

 # purpose : process mouse click at the client side.

 # input   : x, y - mouse coordinates (0 .. 1)

 # output  : none

 #------------------------------------------------------------------

 
 DegasAgent public mouse_click_callback {x y} {
     $self instvar dali_interp_
     $dali_interp_ mouse_click_callback $x $y
 }
 
 DegasAgent instproc print_mem_usage { res_mgr name } {
     puts "Memory used by $name is [$res_mgr get_mem_used $name] bytes"
     after 5000 "$self print_mem_usage $res_mgr $name"
 }
 
 DegasAgent instproc log_cpu_usage { res_mgr } {
     $res_mgr log_cpu_usage
     after 1000 "$self log_cpu_usage $res_mgr"
 }
 
 DegasAgent instproc log { msg } {
     puts stderr "DegasAgent $self : $msg"
 }