Skip to content

CE-178 oap-notification ( oap-notification-mqtt ) #517

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nofateg merged 23 commits into from
Jul 22, 2026
Merged

CE-178 oap-notification ( oap-notification-mqtt ) #517

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
68c8d7f
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
65c7d5e
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
82d8059
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
1fb943b
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
2f15d67
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
de9487a
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
6684d3f
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
e088e1c
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
55c8ce0
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
0c8d244
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
1c2db8c
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
f35b8f2
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
c38b986
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
e2391d5
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
0420787
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
d36f296
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
befb804
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
e8e387d
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
0298b23
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
a23bcc9
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
018e14b
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
2179d60
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 21, 2026
af71ccc
CE-178 oap-notification ( oap-notification-mqtt )
nofateg Jul 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions oap-application/oap-application/src/main/java/oap/application/Kernel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -479,7 +479,7 @@ private void startService( Supervisor supervisor, ModuleItem.ServiceItem si ) {
Service service = si.service;
Object instance = si.instance;
if( service.supervision.supervise ) {
supervisor.startSupervised( si.serviceName, instance,
supervisor.startSupervised( si, instance,
service.supervision.preStartWith,
service.supervision.startWith,
service.supervision.preStopWith,
Expand All @@ -488,13 +488,13 @@ private void startService( Supervisor supervisor, ModuleItem.ServiceItem si ) {
}

if( service.supervision.thread ) {
supervisor.startThread( si.serviceName, instance, applicationConfiguration.shutdown );
supervisor.startThread( si, instance, applicationConfiguration.shutdown );
} else {
if( service.supervision.schedule && service.supervision.cron != null )
supervisor.scheduleCron( si.serviceName, ( Runnable ) instance,
supervisor.scheduleCron( si, ( Runnable ) instance,
service.supervision.cron );
else if( service.supervision.schedule && service.supervision.delay != 0 )
supervisor.scheduleWithFixedDelay( si.serviceName, ( Runnable ) instance,
supervisor.scheduleWithFixedDelay( si, ( Runnable ) instance,
service.supervision.delay, MILLISECONDS );
}
}
Expand Down
8 changes: 4 additions & 4 deletions oap-application/oap-application/src/main/java/oap/application/ModuleItem.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ public boolean equals( Object o ) {
if( this == o ) return true;
if( o == null || getClass() != o.getClass() ) return false;

var that = ( ModuleItem ) o;
ModuleItem that = ( ModuleItem ) o;

return module.name.equals( that.module.name );
}
Expand Down Expand Up @@ -158,7 +158,7 @@ public boolean equals( Object o ) {
if( this == o ) return true;
if( o == null || getClass() != o.getClass() ) return false;

var that = ( ServiceItem ) o;
ServiceItem that = ( ServiceItem ) o;

if( !moduleItem.module.name.equals( that.moduleItem.module.name ) ) return false;
return serviceName.equals( that.serviceName );
Expand All @@ -172,7 +172,7 @@ public int hashCode() {
}

public void addDependsOn( ServiceReference serviceReference ) {
var found = Lists.find2( dependsOn, d -> d.equals( serviceReference ) );
ServiceReference found = Lists.find2( dependsOn, d -> d.equals( serviceReference ) );
if( found == null || found.required ) {
if( found != null ) dependsOn.remove( found );
dependsOn.add( serviceReference );
Expand Down Expand Up @@ -201,7 +201,7 @@ public boolean equals( Object o ) {
if( this == o ) return true;
if( o == null || getClass() != o.getClass() ) return false;

var that = ( ServiceReference ) o;
ServiceReference that = ( ServiceReference ) o;

return serviceItem.serviceName.equals( that.serviceItem.serviceName );
}
Expand Down
115 changes: 58 additions & 57 deletions oap-application/oap-application/src/main/java/oap/application/supervision/Supervisor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import lombok.extern.slf4j.Slf4j;
import oap.application.ApplicationConfiguration;
import oap.application.KernelHelper;
import oap.application.ModuleItem;
import oap.concurrent.Executors;
import oap.util.BiStream;
import oap.util.Dates;
Expand All @@ -44,8 +45,8 @@

@Slf4j
public class Supervisor {
private final LinkedHashMap<String, StartableService> supervised = new LinkedHashMap<>();
private final LinkedHashMap<String, WrapperService<?>> wrappers = new LinkedHashMap<>();
private final LinkedHashMap<ModuleItem.ServiceItem, StartableService> supervised = new LinkedHashMap<>();
private final LinkedHashMap<ModuleItem.ServiceItem, WrapperService<?>> wrappers = new LinkedHashMap<>();

private boolean stopped = false;

Expand Down Expand Up @@ -82,34 +83,34 @@ private static void runAndDetectTimeout( String name, ShutdownConfiguration shut
}
}

public synchronized void startSupervised( String name, Object service,
public synchronized void startSupervised( ModuleItem.ServiceItem si, Object service,
List<String> preStartWith, List<String> startWith,
List<String> preStopWith, List<String> stopWith ) {
this.supervised.put( name, new StartableService( service, preStartWith, startWith, preStopWith, stopWith ) );
this.supervised.put( si, new StartableService( service, preStartWith, startWith, preStopWith, stopWith ) );
}

// public synchronized void startScheduledThread( String name, Object instance, long delay, TimeUnit milliseconds ) {
// this.wrappers.put( name, new ThreadService( name, ( Runnable ) instance, this ) );
// }

public synchronized void startThread( String name, Object instance, ApplicationConfiguration.ModuleShutdown shutdown ) {
this.wrappers.put( name, new ThreadService( name, ( Runnable ) instance, this, shutdown ) );
public synchronized void startThread( ModuleItem.ServiceItem si, Object instance, ApplicationConfiguration.ModuleShutdown shutdown ) {
this.wrappers.put( si, new ThreadService( si.toString(), ( Runnable ) instance, this, shutdown ) );
}

public synchronized void scheduleWithFixedDelay( String name, Runnable service, long delay, TimeUnit unit ) {
this.wrappers.put( name, new DelayScheduledService( service, delay, unit ) );
public synchronized void scheduleWithFixedDelay( ModuleItem.ServiceItem si, Runnable service, long delay, TimeUnit unit ) {
this.wrappers.put( si, new DelayScheduledService( service, delay, unit ) );
}

public synchronized void scheduleCron( String name, Runnable service, String cron ) {
this.wrappers.put( name, new CronScheduledService( service, cron ) );
public synchronized void scheduleCron( ModuleItem.ServiceItem si, Runnable service, String cron ) {
this.wrappers.put( si, new CronScheduledService( service, cron ) );
}

public synchronized void preStart() {
log.debug( "pre starting..." );

this.supervised.forEach( ( name, service ) -> {
log.debug( "pre starting {}...", name );
KernelHelper.setThreadNameSuffix( name );
this.supervised.forEach( ( si, service ) -> {
log.debug( "pre starting {}...", si );
KernelHelper.setThreadNameSuffix( si.toString() );
try {
service.preStart();
} finally {
Expand All @@ -119,45 +120,45 @@ public synchronized void preStart() {

BiStream.of( this.wrappers )
.reversed()
.forEach( ( name, service ) -> {
log.debug( "[{}] pre starting {}...", service.type(), name );
KernelHelper.setThreadNameSuffix( name );
.forEach( ( si, service ) -> {
log.debug( "[{}] pre starting {}...", service.type(), si );
KernelHelper.setThreadNameSuffix( si.toString() );
try {
service.preStart();
} finally {
KernelHelper.restoreThreadName();
}
log.debug( "[{}] pre starting {}... Done.", service.type(), name );
log.debug( "[{}] pre starting {}... Done.", service.type(), si );
} );
}

public synchronized void start() {
log.debug( "starting..." );
this.stopped = false;
this.supervised.forEach( ( name, service ) -> {
log.debug( "starting {}...", name );
this.supervised.forEach( ( si, service ) -> {
log.debug( "starting {}...", si );
long start = System.currentTimeMillis();
KernelHelper.setThreadNameSuffix( name );
KernelHelper.setThreadNameSuffix( si.toString() );
try {
service.start();
} finally {
KernelHelper.restoreThreadName();
}
long end = System.currentTimeMillis();
log.debug( "starting {}... Done. ({}ms)", name, end - start );
log.debug( "starting {}... Done. ({}ms)", si, end - start );
} );

this.wrappers.forEach( ( name, service ) -> {
log.debug( "[{}] starting {}...", service.type(), name );
this.wrappers.forEach( ( si, service ) -> {
log.debug( "[{}] starting {}...", service.type(), si );
long start = System.currentTimeMillis();
KernelHelper.setThreadNameSuffix( name );
KernelHelper.setThreadNameSuffix( si.toString() );
try {
service.start();
} finally {
KernelHelper.restoreThreadName();
}
long end = System.currentTimeMillis();
log.debug( "[{}] starting {}... Done. ({}ms)", service.type(), name, end - start );
log.debug( "[{}] starting {}... Done. ({}ms)", service.type(), si, end - start );
} );
}

Expand All @@ -168,36 +169,36 @@ public synchronized void preStop( ApplicationConfiguration.ModuleShutdown shutdo

BiStream.of( this.wrappers )
.reversed()
.forEach( ( name, service ) -> {
.forEach( ( si, service ) -> {
Runnable func = () -> {
log.debug( "[{}] pre stopping {}...", service.type(), name );
KernelHelper.setThreadNameSuffix( name );
log.debug( "[{}] pre stopping {}...", service.type(), si );
KernelHelper.setThreadNameSuffix( si.toString() );
try {
service.preStop();
} finally {
KernelHelper.restoreThreadName();
}
log.debug( "[{}] pre stopping {}... Done.", service.type(), name );
log.debug( "[{}] pre stopping {}... Done.", service.type(), si );
};

runAndDetectTimeout( name, shutdownConfiguration, func );
runAndDetectTimeout( si.toString(), shutdownConfiguration, func );
} );

BiStream.of( this.supervised )
.reversed()
.forEach( ( name, service ) -> {
.forEach( ( si, service ) -> {
Runnable func = () -> {
log.debug( "pre stopping {}...", name );
KernelHelper.setThreadNameSuffix( name );
log.debug( "pre stopping {}...", si );
KernelHelper.setThreadNameSuffix( si.toString() );
try {
service.preStop();
} finally {
KernelHelper.restoreThreadName();
}
log.debug( "pre stopping {}... Done.", name );
log.debug( "pre stopping {}... Done.", si );
};

runAndDetectTimeout( name, shutdownConfiguration, func );
runAndDetectTimeout( si.toString(), shutdownConfiguration, func );
} );
}
}
Expand All @@ -211,37 +212,37 @@ public synchronized void stop( ApplicationConfiguration.ModuleShutdown shutdown

BiStream.of( this.wrappers )
.reversed()
.forEach( ( name, service ) -> {
.forEach( ( si, service ) -> {
Runnable func = () -> {
log.debug( "[{}] stopping {}...", service.type(), name );
KernelHelper.setThreadNameSuffix( name );
log.debug( "[{}] stopping {}...", service.type(), si );
KernelHelper.setThreadNameSuffix( si.toString() );
try {
service.stop();
} finally {
KernelHelper.restoreThreadName();
}
log.debug( "[{}] stopping {}... Done.", service.type(), name );
log.debug( "[{}] stopping {}... Done.", service.type(), si );
};

runAndDetectTimeout( name, shutdownConfiguration, func );
runAndDetectTimeout( si.toString(), shutdownConfiguration, func );
} );
this.wrappers.clear();

BiStream.of( this.supervised )
.reversed()
.forEach( ( name, service ) -> {
.forEach( ( si, service ) -> {
Runnable func = () -> {
log.debug( "stopping {}...", name );
KernelHelper.setThreadNameSuffix( name );
log.debug( "stopping {}...", si );
KernelHelper.setThreadNameSuffix( si.toString() );
try {
service.stop();
} finally {
KernelHelper.restoreThreadName();
}
log.debug( "stopping {}... Done.", name );
log.debug( "stopping {}... Done.", si );
};

runAndDetectTimeout( name, shutdownConfiguration, func );
runAndDetectTimeout( si.toString(), shutdownConfiguration, func );
} );
this.supervised.clear();
}
Expand All @@ -255,40 +256,40 @@ public synchronized void stop( String serviceName, ApplicationConfiguration.Modu

try( ShutdownConfiguration shutdownConfiguration = new ShutdownConfiguration( shutdown ) ) {
BiStream.of( this.wrappers )
.filter( ( name, _ ) -> name.equals( serviceName ) )
.forEach( ( name, service ) -> {
.filter( ( si, _ ) -> si.serviceName.equals( serviceName ) )
.forEach( ( si, service ) -> {
Runnable func = () -> {
log.debug( "[{}] stopping {}...", service.type(), name );
KernelHelper.setThreadNameSuffix( name );
log.debug( "[{}] stopping {}...", service.type(), si );
KernelHelper.setThreadNameSuffix( si.toString() );
try {
service.preStop();
service.stop();
} finally {
KernelHelper.restoreThreadName();
}
log.debug( "[{}] stopping {}... Done.", service.type(), name );
log.debug( "[{}] stopping {}... Done.", service.type(), si );
};

runAndDetectTimeout( name, shutdownConfiguration, func );
runAndDetectTimeout( si.toString(), shutdownConfiguration, func );
} );
this.wrappers.clear();

BiStream.of( this.supervised )
.filter( ( name, _ ) -> name.equals( serviceName ) )
.forEach( ( name, service ) -> {
.filter( ( si, _ ) -> si.serviceName.equals( serviceName ) )
.forEach( ( si, service ) -> {
Runnable func = () -> {
log.debug( "stopping {}...", name );
KernelHelper.setThreadNameSuffix( name );
log.debug( "stopping {}...", si );
KernelHelper.setThreadNameSuffix( si.toString() );
try {
service.preStop();
service.stop();
} finally {
KernelHelper.restoreThreadName();
}
log.debug( "stopping {}... Done.", name );
log.debug( "stopping {}... Done.", si );
};

runAndDetectTimeout( name, shutdownConfiguration, func );
runAndDetectTimeout( si.toString(), shutdownConfiguration, func );
} );
}
}
Expand Down
6 changes: 3 additions & 3 deletions ...ap-logstream/oap-logstream-test/src/test/java/oap/logstream/disk/RowBinaryWriterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,7 @@ public void testWrite() throws IOException {
LogId logId = new LogId( "", "log", "log",
Map.of( "p", "1" ), headers, types );
Path logs = testDirectoryFixture.testPath( "logs" );
try( RowBinaryWriter writer = new RowBinaryWriter( templateEngineFixture.templateEngine, logs, FILE_PATTERN, logId, 1024, BPH_12, 20, "localhost" ) ) {
try( RowBinaryWriter writer = new RowBinaryWriter( templateEngineFixture.templateEngine, logs, FILE_PATTERN, logId, 1024, BPH_12, 20, "localhost", null ) ) {
writer.write( CURRENT_PROTOCOL_VERSION, content1 );
writer.write( CURRENT_PROTOCOL_VERSION, content2 );
}
Expand Down Expand Up @@ -124,7 +124,7 @@ public void testConcurrency() throws IOException {

int count = 10;

try( RowBinaryWriter writer = new RowBinaryWriter( templateEngineFixture.templateEngine, logs, FILE_PATTERN, logId, 1024, BPH_12, 20, "localhost" ) ) {
try( RowBinaryWriter writer = new RowBinaryWriter( templateEngineFixture.templateEngine, logs, FILE_PATTERN, logId, 1024, BPH_12, 20, "localhost", null ) ) {
try( ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor() ) {
for( long i = 0; i < count; i++ ) {

Expand Down Expand Up @@ -168,7 +168,7 @@ public void testWriteToNewVersionWhenCompleted() throws IOException {
Path v1 = logs.resolve( "1-file-02-47b82ddc0-1.rb.gz.rb.gz" );
Path v2 = logs.resolve( "1-file-02-47b82ddc0-2.rb.gz.rb.gz" );

try( RowBinaryWriter writer = new RowBinaryWriter( templateEngineFixture.templateEngine, logs, FILE_PATTERN, logId, 1024, BPH_12, 20, "localhost" ) ) {
try( RowBinaryWriter writer = new RowBinaryWriter( templateEngineFixture.templateEngine, logs, FILE_PATTERN, logId, 1024, BPH_12, 20, "localhost", null ) ) {
writer.write( CURRENT_PROTOCOL_VERSION, content1 );

writer.refresh();
Expand Down
Loading
Loading