@@ -0,0 +1,524 @@
# include <stdio.h>
# include <string.h>
# include "../../lib/debug_config.h"
# include "../../lib/mem.h"
# include "../../lib/platform_compat.h"
# include "../../lib/u_async.h"
# include "../utun_instance.h"
# include "../chat/chat_core.h"
# include "stream_member.h"
# include "stream_source.h"
# include "stream_priv.h"
struct stream_session {
struct stream_session * next ;
struct stream_service * service ;
struct stream_descriptor descriptor ;
uint8_t public_key [ SC_PUBKEY_SIZE ] ;
struct stream_source * source ;
struct stream_buffer * buffer ;
uint8_t * output ;
void * delivery ;
uint64_t next_seq ;
unsigned owners , holds ;
int anchored , closing , invalidated , finished , end_delivered , reason ;
} ;
struct STREAM_HANDLE {
struct STREAM_HANDLE * next ;
struct stream_service * service ;
struct stream_session * session ;
int publisher ;
int reason ;
} ;
struct stream_service {
struct UTUN_INSTANCE * inst ;
struct stream_session * sessions ;
struct STREAM_HANDLE * handles ;
stream_event_fn callback ;
void * arg ;
void * reconcile ;
int stopping , dispatching , registered , initializing ;
} ;
static void session_deliver ( void * arg ) ;
static void service_reconcile ( void * arg ) ;
/* Активность CHAT проверяется до чтения профиля, выделения 32 МБ и создания handles. */
static struct TOPO_GROUP * stream_group ( struct UTUN_INSTANCE * inst , uint64_t id ) {
struct TOPO_GROUP * group = inst & & inst - > topo_groups ? topo_groups_find ( inst - > topo_groups , id ) : NULL ;
return group & & group - > group_type = = TOPO_GROUP_TYPE_CHAT & & ! group - > stopping ? group : NULL ;
}
/* Один локальный экземпляр на источник группы; закрывающийся старый запуск не мешает новому. */
static struct stream_session * session_find ( struct stream_service * service , uint64_t group_id , uint64_t source_id ) {
for ( struct stream_session * s = service - > sessions ; s ; s = s - > next )
if ( ! s - > closing & & ! s - > invalidated & & s - > descriptor . group_id = = group_id & & s - > descriptor . source_node_id = = source_id ) return s ;
return NULL ;
}
/* Две опции заменяются одной подписанной записью; список отражает экземпляры, не отдельные handles. */
static int service_publish_options ( struct stream_service * service , struct TOPO_GROUP * group ) {
if ( ! group ) { DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream options group missing " ) ; return STREAM_ERR_STATE ; }
struct stream_member_state state = { 0 } ;
for ( struct stream_session * s = service - > sessions ; s ; s = s - > next ) {
if ( s - > closing | | s - > invalidated | | s - > descriptor . group_id ! = group - > group_id ) continue ;
if ( state . count = = STREAM_MEMBER_MAX_SUBSCRIPTIONS ) {
DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " local subscription list exceeds member limit group=%llx " ,
( unsigned long long ) group - > group_id ) ; return STREAM_ERR_ARGUMENT ;
}
struct stream_subscription_key entry = { s - > descriptor . source_node_id , s - > descriptor . source_epoch } ;
size_t at = state . count + + ;
while ( at & & ( state . subscriptions [ at - 1 ] . source_node_id > entry . source_node_id
| | ( state . subscriptions [ at - 1 ] . source_node_id = = entry . source_node_id
& & state . subscriptions [ at - 1 ] . source_epoch > entry . source_epoch ) ) ) {
state . subscriptions [ at ] = state . subscriptions [ at - 1 ] ; at - - ;
}
state . subscriptions [ at ] = entry ;
if ( s - > source & & ! s - > finished ) { state . source = s - > descriptor ; state . published = 1 ; }
}
char source [ MS_USERINFO_MAX + 1 ] , subscriptions [ MS_USERINFO_MAX + 1 ] ;
int result = stream_member_format ( & state , source , subscriptions ) ;
if ( result ! = STREAM_OK ) return result ;
struct chat_profile_option options [ ] = {
{ STREAM_MEMBER_SOURCE , source [ 0 ] ? source : NULL } ,
{ STREAM_MEMBER_SUBSCRIPTIONS , subscriptions [ 0 ] ? subscriptions : NULL }
} ;
if ( chat_core_set_my_profile_options ( service - > inst , group - > channel_id , options , 2 ) < 0 ) {
DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " local member stream options update failed group=%llx owners=%zu " ,
( unsigned long long ) group - > group_id , state . count ) ; return STREAM_ERR_STATE ;
}
DEBUG_INFO ( DEBUG_CATEGORY_STREAM , " local stream options group=%llx published=%d subscriptions=%zu " ,
( unsigned long long ) group - > group_id , state . published , state . count ) ;
return STREAM_OK ;
}
/* Ресурсы живут до возврата пользовательского callback/write, даже если он закрыл последний handle. */
static void session_free ( struct stream_session * s ) {
struct stream_session * * link = & s - > service - > sessions ;
while ( * link ! = s ) link = & ( * link ) - > next ;
* link = s - > next ;
if ( s - > delivery ) uasync_call_soon_cancel ( s - > service - > inst - > ua , s - > delivery ) ;
if ( s - > source ) stream_source_destroy ( s - > source ) ;
else stream_buffer_destroy ( s - > buffer ) ;
u_free ( s - > output ) ;
DEBUG_INFO ( DEBUG_CATEGORY_STREAM , " local stream released group=%llx source=%llx epoch=%llx reason=%d " ,
( unsigned long long ) s - > descriptor . group_id , ( unsigned long long ) s - > descriptor . source_node_id ,
( unsigned long long ) s - > descriptor . source_epoch , s - > reason ) ;
u_free ( s ) ;
}
/* Delivery выполняется отдельным callback, а
static void session_schedule ( struct stream_session * s ) {
if ( s - > service - > stopping | | s - > delivery ) return ;
s - > delivery = uasync_call_soon ( s - > service - > inst - > ua , s , session_deliver ) ;
if ( ! s - > delivery ) DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream delivery scheduling failed source=%llx " ,
( unsigned long long ) s - > descriptor . source_node_id ) ;
}
/* Снять локальные ссылки и публикацию; CLOSED также отложен и отменяется при stop всего сервиса. */
static void session_close ( struct stream_session * s , int reason , int update_profile , int notify ) {
if ( s - > closing ) return ;
s - > closing = 1 ; s - > reason = reason ; s - > owners = 0 ;
for ( struct STREAM_HANDLE * h = s - > service - > handles ; h ; h = h - > next )
if ( h - > session = = s ) { h - > session = NULL ; h - > reason = reason ; }
if ( s - > delivery ) { uasync_call_soon_cancel ( s - > service - > inst - > ua , s - > delivery ) ; s - > delivery = NULL ; }
struct TOPO_GROUP * group = s - > service - > inst - > topo_groups
? topo_groups_find ( s - > service - > inst - > topo_groups , s - > descriptor . group_id ) : NULL ;
if ( update_profile & & group & & service_publish_options ( s - > service , group ) ! = STREAM_OK )
DEBUG_WARN ( DEBUG_CATEGORY_STREAM , " stream closed with stale own member options group=%llx " ,
( unsigned long long ) s - > descriptor . group_id ) ;
if ( notify & & s - > service - > callback & & ! s - > service - > stopping ) session_schedule ( s ) ;
else if ( ! s - > holds ) session_free ( s ) ;
}
/* Уведомление использует собственный scratch экземпляра; close внутри callback не освобождает его преждевременно. */
static void session_deliver ( void * arg ) {
struct stream_session * s = arg ;
struct stream_service * service = s - > service ;
s - > delivery = NULL ; s - > holds + + ;
struct stream_event event = { . descriptor = s - > descriptor } ;
int emit = 0 , again = 0 ;
if ( s - > closing ) { event . type = STREAM_EVENT_CLOSED ; event . reason = s - > reason ; emit = 1 ; }
else if ( s - > invalidated ) {
DEBUG_DEBUG ( DEBUG_CATEGORY_STREAM , " invalidated stream awaits deferred close source=%llx " ,
( unsigned long long ) s - > descriptor . source_node_id ) ;
} else if ( s - > anchored ) {
struct stream_buffer_stats stats ;
stream_buffer_get_stats ( s - > buffer , & stats ) ;
size_t length = 0 ;
int result = s - > next_seq = = UINT64_MAX ? STREAM_WAIT
: stream_buffer_decode ( s - > buffer , s - > next_seq , s - > output , stream_input_size ( & s - > descriptor ) , & length ) ;
if ( result = = STREAM_OK ) {
event . type = STREAM_EVENT_DATA ; event . first_seq = event . last_seq = s - > next_seq + + ;
event . data = s - > output ; event . length = length ; emit = 1 ; again = 1 ;
} else if ( result = = STREAM_EXPIRED ) {
event . type = STREAM_EVENT_GAP ; event . first_seq = s - > next_seq ; event . last_seq = stats . expired_before - 1 ;
s - > next_seq = stats . expired_before ; emit = 1 ; again = 1 ;
} else if ( result < 0 ) {
session_close ( s , STREAM_CLOSE_FAILED , 1 , 1 ) ;
} else if ( s - > finished & & s - > next_seq > stats . highest_seq & & ! s - > end_delivered ) {
event . type = STREAM_EVENT_END ; event . first_seq = event . last_seq = stats . highest_seq ;
s - > end_delivered = 1 ; emit = 1 ;
}
} else if ( s - > finished & & ! s - > end_delivered ) {
event . type = STREAM_EVENT_END ; event . first_seq = event . last_seq = UINT64_MAX ;
s - > end_delivered = 1 ; emit = 1 ;
}
if ( emit & & service - > callback & & ! service - > stopping ) {
service - > dispatching + + ;
service - > callback ( service - > inst , & event , service - > arg ) ;
service - > dispatching - - ;
}
s - > holds - - ;
if ( s - > closing ) { if ( ! s - > holds & & ! s - > delivery ) session_free ( s ) ; }
else if ( again ) session_schedule ( s ) ;
}
/* Готовая source-пачка уже полностью в кольце. Пользовательский callback здесь не вызывается. */
static void source_ready ( struct stream_source * source , uint64_t seq , void * arg ) {
( void ) source ;
struct stream_session * s = arg ;
if ( s - > closing ) return ;
if ( ! s - > anchored ) { s - > anchored = 1 ; s - > next_seq = seq ; }
session_schedule ( s ) ;
}
/* Отложенная сверка не разрушает экземпляры из BGP/member callbacks. */
static void service_reconcile ( void * arg ) {
struct stream_service * service = arg ;
service - > reconcile = NULL ;
struct stream_session * snapshot = service - > sessions ;
for ( struct stream_session * s = snapshot ; s ; s = s - > next ) s - > holds + + ;
struct stream_session * s = snapshot ;
while ( s ) {
struct stream_session * next = s - > next ;
if ( ! s - > closing & & s - > invalidated ) session_close ( s , s - > reason , 1 , 1 ) ;
else if ( ! s - > closing & & ! s - > source ) {
struct stream_member_state state ; uint8_t key [ 32 ] ;
int result = stream_member_load ( service - > inst , s - > descriptor . group_id , s - > descriptor . source_node_id , & state , key ) ;
if ( result = = STREAM_ERR_NOT_MEMBER | | result = = STREAM_ERR_SIGNATURE | | result = = STREAM_ERR_FORMAT )
session_close ( s , STREAM_CLOSE_MEMBER , 1 , 1 ) ;
else if ( result = = STREAM_OK & & ( memcmp ( key , s - > public_key , 32 )
| | ( state . published & & ( state . source . source_epoch ! = s - > descriptor . source_epoch
| | state . source . n ! = s - > descriptor . n | | state . source . r ! = s - > descriptor . r
| | state . source . block_size ! = s - > descriptor . block_size ) ) ) )
session_close ( s , STREAM_CLOSE_SOURCE_CHANGED , 1 , 1 ) ;
}
s = next ;
}
s = snapshot ;
while ( s ) {
struct stream_session * next = s - > next ;
s - > holds - - ;
if ( s - > closing & & ! s - > holds & & ! s - > delivery ) session_free ( s ) ;
s = next ;
}
}
/* Общий props сигнал относится и к Merkle commit, и к BGP. Payload adm_tags здесь не используется. */
static void member_changed ( uint64_t node_id , const char * tags , const char * channel , void * arg ) {
( void ) tags ;
struct stream_service * service = arg ;
if ( service - > stopping | | ! service - > sessions ) return ;
for ( struct stream_session * s = service - > sessions ; s ; s = s - > next ) {
if ( s - > closing | | s - > source | | s - > descriptor . source_node_id ! = node_id ) continue ;
struct TOPO_GROUP * group = stream_group ( service - > inst , s - > descriptor . group_id ) ;
if ( ! group | | ! channel | | strcmp ( group - > channel_id , channel ) ) continue ;
struct stream_member_state state ; uint8_t key [ 32 ] ;
int result = stream_member_load ( service - > inst , group - > group_id , node_id , & state , key ) ;
if ( result = = STREAM_ERR_NOT_MEMBER | | result = = STREAM_ERR_SIGNATURE | | result = = STREAM_ERR_FORMAT ) {
s - > invalidated = 1 ; s - > reason = STREAM_CLOSE_MEMBER ;
} else if ( result = = STREAM_OK & & ( memcmp ( key , s - > public_key , 32 )
| | ( state . published & & ( state . source . source_epoch ! = s - > descriptor . source_epoch
| | state . source . n ! = s - > descriptor . n | | state . source . r ! = s - > descriptor . r
| | state . source . block_size ! = s - > descriptor . block_size ) ) ) ) {
s - > invalidated = 1 ; s - > reason = STREAM_CLOSE_SOURCE_CHANGED ;
}
}
if ( service - > reconcile ) return ;
service - > reconcile = uasync_call_soon ( service - > inst - > ua , service , service_reconcile ) ;
if ( ! service - > reconcile ) DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream member reconciliation scheduling failed " ) ;
}
/* Снять старые собственные опции после аварии; неполный/отсутствующий профиль не является активной подпиской. */
int stream_service_init ( struct UTUN_INSTANCE * inst ) {
if ( ! inst | | ! inst - > core_started | | ! chat_core_get_db ( inst ) | | ! inst - > topo_groups ) {
DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream init requires core and chat DB " ) ; return STREAM_ERR_STATE ;
}
if ( inst - > stream ) return STREAM_OK ;
struct stream_service * service = u_calloc ( 1 , sizeof ( * service ) ) ;
if ( ! service ) { DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream service allocation failed " ) ; return STREAM_ERR_MEMORY ; }
service - > inst = inst ; service - > initializing = 1 ; inst - > stream = service ;
for ( struct ll_entry * e = inst - > topo_groups - > group_list - > head ; e ; e = e - > next ) {
struct TOPO_GROUP * group = ( struct TOPO_GROUP * ) e ;
if ( group - > group_type ! = TOPO_GROUP_TYPE_CHAT ) continue ;
struct stream_member_state state ;
int result = stream_member_load ( inst , group - > group_id , inst - > node_id , & state , NULL ) ;
if ( result = = STREAM_OK & & ! state . published & & ! state . count ) continue ;
if ( result = = STREAM_WAIT | | result = = STREAM_ERR_NOT_MEMBER ) continue ;
if ( service_publish_options ( service , group ) ! = STREAM_OK ) {
service - > initializing = 0 ; stream_service_destroy ( inst ) ; return STREAM_ERR_STATE ;
}
}
if ( member_sync_add_props_cbk ( inst , member_changed , service ) < 0 ) {
service - > initializing = 0 ; stream_service_destroy ( inst ) ; return STREAM_ERR_MEMORY ;
}
service - > registered = 1 ; service - > initializing = 0 ;
DEBUG_INFO ( DEBUG_CATEGORY_STREAM , " stream service initialized node=%llx " , ( unsigned long long ) inst - > node_id ) ;
return STREAM_OK ;
}
/* Detached handles принадлежат callers и остаются безопасными для close после остановки чата/ядра. */
void stream_service_destroy ( struct UTUN_INSTANCE * inst ) {
struct stream_service * service = inst ? inst - > stream : NULL ;
if ( ! service ) return ;
if ( service - > dispatching | | service - > initializing ) {
DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream service destroy inside callback/init rejected " ) ; return ;
}
for ( struct stream_session * s = service - > sessions ; s ; s = s - > next )
if ( s - > holds ) { DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream stop during active operation rejected " ) ; return ; }
service - > stopping = 1 ;
if ( service - > registered ) member_sync_remove_props_cbk ( inst , member_changed , service ) ;
if ( service - > reconcile ) uasync_call_soon_cancel ( inst - > ua , service - > reconcile ) ;
while ( service - > sessions ) {
struct stream_session * s = service - > sessions ;
if ( ! s - > closing ) session_close ( s , STREAM_CLOSE_SERVICE , 1 , 0 ) ;
else session_free ( s ) ;
}
for ( struct STREAM_HANDLE * h = service - > handles ; h ; h = h - > next ) { h - > service = NULL ; h - > session = NULL ; h - > reason = STREAM_CLOSE_SERVICE ; }
inst - > stream = NULL ;
DEBUG_INFO ( DEBUG_CATEGORY_STREAM , " stream service stopped node=%llx " , ( unsigned long long ) inst - > node_id ) ;
u_free ( service ) ;
}
void stream_cancel_group ( struct UTUN_INSTANCE * inst , uint64_t group_id ) {
struct stream_service * service = inst ? inst - > stream : NULL ;
if ( ! service ) return ;
struct stream_session * s = service - > sessions ;
while ( s ) {
struct stream_session * next = s - > next ;
if ( ! s - > closing & & s - > descriptor . group_id = = group_id ) session_close ( s , STREAM_CLOSE_GROUP , 1 , 0 ) ;
s = next ;
}
}
void stream_set_event_callback ( struct UTUN_INSTANCE * inst , stream_event_fn callback , void * arg ) {
if ( ! inst | | ! inst - > stream | | inst - > stream - > stopping ) { DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream callback before init " ) ; return ; }
inst - > stream - > callback = callback ; inst - > stream - > arg = arg ;
if ( callback ) for ( struct stream_session * s = inst - > stream - > sessions ; s ; s = s - > next )
if ( ! s - > closing ) session_schedule ( s ) ;
}
/* Подготовить все ресурсы до публикации member-опций; rollback не оставляет видимую подписку. */
static int session_open ( struct UTUN_INSTANCE * inst , const struct stream_descriptor * d , const uint8_t key [ 32 ] , int publisher ,
struct STREAM_HANDLE * * out ) {
struct stream_service * service = inst - > stream ;
struct stream_session * s = session_find ( service , d - > group_id , d - > source_node_id ) ;
int created = s = = NULL ;
if ( s & & ( publisher | | s - > descriptor . source_epoch ! = d - > source_epoch ) ) {
DEBUG_WARN ( DEBUG_CATEGORY_STREAM , " stream source already owned with another publication/epoch " ) ; return STREAM_ERR_STATE ;
}
struct STREAM_HANDLE * h = u_calloc ( 1 , sizeof ( * h ) ) ;
if ( ! h ) { DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream handle allocation failed " ) ; return STREAM_ERR_MEMORY ; }
if ( created ) {
s = u_calloc ( 1 , sizeof ( * s ) ) ;
if ( ! s ) { DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream session allocation failed " ) ; u_free ( h ) ; return STREAM_ERR_MEMORY ; }
s - > service = service ; s - > descriptor = * d ; memcpy ( s - > public_key , key , 32 ) ;
s - > output = u_malloc ( ( uint32_t ) stream_input_size ( d ) ) ;
int result = s - > output ? STREAM_OK : STREAM_ERR_MEMORY ;
if ( result = = STREAM_OK & & publisher ) {
result = stream_source_create ( & s - > source , inst - > ua , d , inst - > my_ed25519_privkey , source_ready , s ) ;
if ( result = = STREAM_OK ) s - > buffer = stream_source_buffer ( s - > source ) ;
} else if ( result = = STREAM_OK ) result = stream_buffer_create ( & s - > buffer , d , key , STREAM_BUFFER_BYTES ) ;
s - > next = service - > sessions ; service - > sessions = s ;
if ( result ! = STREAM_OK ) {
DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream session initialization failed result=%d " , result ) ;
session_free ( s ) ; u_free ( h ) ; return result ;
}
}
h - > service = service ; h - > session = s ; h - > publisher = publisher ;
h - > next = service - > handles ; service - > handles = h ; s - > owners + + ;
s - > holds + + ;
int result = created ? service_publish_options ( service , stream_group ( inst , d - > group_id ) ) : STREAM_OK ;
if ( s - > closing & & result = = STREAM_OK ) {
DEBUG_WARN ( DEBUG_CATEGORY_STREAM , " stream open cancelled during member callbacks " ) ; result = STREAM_ERR_STATE ;
}
if ( result ! = STREAM_OK ) {
struct STREAM_HANDLE * * link = & service - > handles ;
while ( * link ! = h ) link = & ( * link ) - > next ;
* link = h - > next ;
if ( ! s - > closing ) { s - > owners - - ; session_close ( s , STREAM_CLOSE_FAILED , 0 , 0 ) ; }
u_free ( h ) ; s - > holds - - ;
if ( ! s - > holds & & ! s - > delivery ) session_free ( s ) ;
return result ;
}
s - > holds - - ;
* out = h ;
DEBUG_INFO ( DEBUG_CATEGORY_STREAM , " local stream handle opened group=%llx source=%llx epoch=%llx publisher=%d owners=%u " ,
( unsigned long long ) d - > group_id , ( unsigned long long ) d - > source_node_id ,
( unsigned long long ) d - > source_epoch , publisher , s - > owners ) ;
return STREAM_OK ;
}
/* Контекст не создаётся для неизвестной группы, неавторизованного self или неподтверждённого ключа. */
static int stream_open_check ( struct UTUN_INSTANCE * inst , uint64_t group_id , struct STREAM_HANDLE * * out ) {
if ( ! out | | * out | | ! inst | | ! inst - > chat_started | | ! inst - > stream | | inst - > stream - > stopping | | ! stream_group ( inst , group_id ) ) {
DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream open invalid state/group=%llx " , ( unsigned long long ) group_id ) ; return STREAM_ERR_STATE ;
}
struct stream_member_state own ; uint8_t key [ 32 ] ;
int result = stream_member_load ( inst , group_id , inst - > node_id , & own , key ) ;
if ( result ! = STREAM_OK ) return result ;
if ( memcmp ( key , inst - > my_ed25519_pubkey , 32 ) ) {
DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream own member key differs from core identity " ) ; return STREAM_ERR_SIGNATURE ;
}
return STREAM_OK ;
}
int stream_publish_open ( struct UTUN_INSTANCE * inst , uint64_t group_id , const struct stream_parameters * parameters ,
struct STREAM_HANDLE * * out ) {
int result = stream_open_check ( inst , group_id , out ) ;
if ( result ! = STREAM_OK ) return result ;
struct stream_descriptor d = { . group_id = group_id , . source_node_id = inst - > node_id ,
. n = parameters ? parameters - > n : 8 , . r = parameters ? parameters - > r : 2 ,
. block_size = parameters ? parameters - > block_size : STREAM_RS_DEFAULT_SIZE } ;
do {
if ( random_bytes ( ( uint8_t * ) & d . source_epoch , sizeof ( d . source_epoch ) ) ! = 0 ) {
DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream source epoch generation failed " ) ; return STREAM_ERR_CRYPTO ;
}
} while ( ! d . source_epoch ) ;
if ( stream_descriptor_validate ( & d ) ! = STREAM_OK ) return STREAM_ERR_ARGUMENT ;
return session_open ( inst , & d , inst - > my_ed25519_pubkey , 1 , out ) ;
}
int stream_subscribe_open ( struct UTUN_INSTANCE * inst , uint64_t group_id , uint64_t source_id , struct STREAM_HANDLE * * out ) {
int result = stream_open_check ( inst , group_id , out ) ;
if ( result ! = STREAM_OK ) return result ;
struct stream_session * s = session_find ( inst - > stream , group_id , source_id ) ;
if ( s ) return session_open ( inst , & s - > descriptor , s - > public_key , 0 , out ) ;
struct stream_member_state state ; uint8_t key [ 32 ] ;
result = stream_member_load ( inst , group_id , source_id , & state , key ) ;
if ( result ! = STREAM_OK ) return result ;
if ( ! state . published ) { DEBUG_DEBUG ( DEBUG_CATEGORY_STREAM , " source not yet published source=%llx " , ( unsigned long long ) source_id ) ; return STREAM_WAIT ; }
return session_open ( inst , & state . source , key , 0 , out ) ;
}
/* Publisher авторитетно останавливает трансляцию; reader снимает только собственное владение. */
void stream_close ( struct STREAM_HANDLE * h ) {
if ( ! h ) return ;
struct stream_service * service = h - > service ;
if ( service ) {
struct STREAM_HANDLE * * link = & service - > handles ;
while ( * link ! = h ) link = & ( * link ) - > next ;
* link = h - > next ;
}
struct stream_session * s = h - > session ;
if ( s & & ( h - > publisher | | - - s - > owners = = 0 ) ) session_close ( s , STREAM_CLOSE_STOPPED , 1 , 1 ) ;
u_free ( h ) ;
}
int stream_handle_active ( const struct STREAM_HANDLE * h ) { return h & & h - > session & & ! h - > session - > closing & & ! h - > session - > invalidated ; }
int stream_handle_reason ( const struct STREAM_HANDLE * h ) {
return h & & h - > session & & h - > session - > invalidated ? h - > session - > reason : h ? h - > reason : STREAM_CLOSE_STOPPED ;
}
int stream_get_info ( const struct STREAM_HANDLE * h , struct stream_info * info ) {
if ( ! info | | ! stream_handle_active ( h ) ) { DEBUG_WARN ( DEBUG_CATEGORY_STREAM , " stream info detached/invalid handle " ) ; return STREAM_ERR_STATE ; }
struct stream_session * s = h - > session ;
struct stream_info result = { . descriptor = s - > descriptor , . owners = s - > owners , . publisher = s - > source ! = NULL , . finished = s - > finished } ;
stream_buffer_get_stats ( s - > buffer , & result . buffer ) ;
* info = result ;
return STREAM_OK ;
}
size_t stream_publish_capacity ( const struct STREAM_HANDLE * h ) {
if ( ! stream_handle_active ( h ) | | ! h - > publisher ) { DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream capacity not a publisher " ) ; return 0 ; }
return stream_source_input_capacity ( h - > session - > source ) ;
}
int stream_publish_write ( struct STREAM_HANDLE * h , const uint8_t * data , size_t length , size_t * accepted ) {
if ( accepted ) * accepted = 0 ;
if ( ! stream_handle_active ( h ) | | ! h - > publisher | | ! accepted ) {
DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream write not an active publisher " ) ; return STREAM_ERR_STATE ;
}
struct stream_session * s = h - > session ; s - > holds + + ;
int result = stream_source_write ( s - > source , data , length , accepted ) ;
if ( result ! = STREAM_OK & & result ! = STREAM_ERR_ARGUMENT ) session_close ( s , STREAM_CLOSE_FAILED , 1 , 1 ) ;
s - > holds - - ;
if ( s - > closing & & ! s - > holds & & ! s - > delivery ) session_free ( s ) ;
return result ;
}
int stream_publish_finish ( struct STREAM_HANDLE * h ) {
if ( ! stream_handle_active ( h ) | | ! h - > publisher ) { DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream finish not a publisher " ) ; return STREAM_ERR_STATE ; }
struct stream_session * s = h - > session ;
if ( s - > finished ) return STREAM_OK ;
s - > holds + + ;
s - > finished = 1 ;
int result = service_publish_options ( s - > service , stream_group ( s - > service - > inst , s - > descriptor . group_id ) ) ;
if ( result ! = STREAM_OK | | s - > closing ) {
if ( ! s - > closing ) s - > finished = 0 ;
s - > holds - - ;
if ( s - > closing & & ! s - > holds & & ! s - > delivery ) session_free ( s ) ;
return result ! = STREAM_OK ? result : STREAM_ERR_STATE ;
}
result = stream_source_finish ( s - > source ) ;
if ( result ! = STREAM_OK ) session_close ( s , STREAM_CLOSE_FAILED , 1 , 1 ) ;
else session_schedule ( s ) ;
s - > holds - - ;
if ( s - > closing & & ! s - > holds & & ! s - > delivery ) session_free ( s ) ;
return result ;
}
/* Сначала только ключ/наличие подписки, затем crypto/FEC. Парсер пути и peer authorization — выше этого входа. */
int stream_ingest_block ( struct UTUN_INSTANCE * inst , const uint8_t * wire , size_t length ) {
if ( ! inst | | ! wire | | length < STREAM_BLOCK_RS_OFF + RS_HEADER_SIZE ) {
DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream ingest invalid length=%zu " , length ) ; return STREAM_ERR_ARGUMENT ;
}
if ( ! inst - > stream | | inst - > stream - > stopping ) return STREAM_WAIT ;
uint64_t group = stream_read_be ( wire + STREAM_BLOCK_GROUP_OFF , 8 ) , source = stream_read_be ( wire + STREAM_BLOCK_SOURCE_OFF , 8 ) ;
struct stream_session * s = session_find ( inst - > stream , group , source ) ;
if ( ! s ) { DEBUG_TRACE ( DEBUG_CATEGORY_STREAM , " unsubscribed stream ignored group=%llx source=%llx " ,
( unsigned long long ) group , ( unsigned long long ) source ) ; return STREAM_WAIT ; }
if ( s - > source ) { DEBUG_WARN ( DEBUG_CATEGORY_STREAM , " incoming DATA to own publisher rejected " ) ; return STREAM_ERR_STATE ; }
int result = stream_buffer_put ( s - > buffer , wire , length ) ;
if ( result = = STREAM_OK ) {
if ( ! s - > anchored ) { s - > anchored = 1 ; s - > next_seq = stream_read_be ( wire + STREAM_BLOCK_SEQ_OFF , 8 ) ; }
session_schedule ( s ) ;
}
return result ;
}
/* Выборка идёт из live BGP-путей группы, не из nodes.online или глобальных соединений. */
int stream_get_candidates ( const struct STREAM_HANDLE * h , struct stream_candidate * * out , size_t * count ) {
if ( ! out | | * out | | ! count | | ! stream_handle_active ( h ) ) {
DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream candidates invalid handle/output " ) ; return STREAM_ERR_ARGUMENT ;
}
struct stream_session * s = h - > session ;
struct UTUN_INSTANCE * inst = s - > service - > inst ;
struct TOPO_GROUP * group = stream_group ( inst , s - > descriptor . group_id ) ;
if ( ! group ) { DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream candidates group missing " ) ; return STREAM_ERR_STATE ; }
size_t capacity = ( size_t ) group - > nodes - > count ;
struct stream_candidate * candidates = capacity ? u_calloc ( ( uint32_t ) capacity , sizeof ( * candidates ) ) : NULL ;
if ( capacity & & ! candidates ) { DEBUG_ERROR ( DEBUG_CATEGORY_STREAM , " stream candidates allocation failed count=%zu " , capacity ) ; return STREAM_ERR_MEMORY ; }
size_t found = 0 ;
for ( struct ll_entry * e = group - > nodes - > head ; e ; e = e - > next ) {
struct TOPO_GROUP_NODE * node = ( struct TOPO_GROUP_NODE * ) e ;
if ( node - > node_id = = inst - > node_id | | ! stream_member_online ( inst , group - > group_id , node - > node_id ) ) continue ;
struct TOPO_NODE * identity = topo_node_registry_find ( inst - > topo_groups , node - > node_id ) ;
int source = node - > node_id = = s - > descriptor . source_node_id ;
if ( identity & & identity - > client_type = = CLIENT_TYPE_MOBILE & & ! source ) continue ;
struct stream_member_state state ;
if ( stream_member_load ( inst , group - > group_id , node - > node_id , & state , NULL ) ! = STREAM_OK
| | ! stream_member_has_subscription ( & state , s - > descriptor . source_node_id , s - > descriptor . source_epoch ) ) continue ;
candidates [ found + + ] = ( struct stream_candidate ) { . node_id = node - > node_id , . source = ( uint8_t ) source ,
. client_type = identity ? identity - > client_type : UINT8_MAX } ;
}
if ( ! found ) { u_free ( candidates ) ; candidates = NULL ; }
* out = candidates ; * count = found ;
DEBUG_DEBUG ( DEBUG_CATEGORY_STREAM , " stream candidates source=%llx eligible=%zu BGP=%zu " ,
( unsigned long long ) s - > descriptor . source_node_id , found , capacity ) ;
return STREAM_OK ;
}
evgeny committed