@stream λ°μ½λ μ΄ν°λ Server-Sent Events(SSE)λ₯Ό μ¬μ©ν μ€μκ° μ€νΈλ¦¬λ° APIλ₯Ό μμ±ν©λλ€.
κΈ°λ³Έ μ¬μ©λ²
import { BaseModelClass , stream } from "sonamu" ;
import { z } from "zod" ;
const NotificationEventsSchema = z . object ({
message: z . object ({
id: z . number (),
text: z . string (),
timestamp: z . date (),
}),
status: z . object ({
type: z . enum ([ "connected" , "disconnected" ]),
userId: z . number (),
}),
});
class NotificationModelClass extends BaseModelClass {
@ stream ({
type: "sse" ,
events: NotificationEventsSchema ,
})
async subscribe ( userId : number ) {
const sse = Sonamu . getContext (). createSSE ( NotificationEventsSchema );
// μ°κ²° μμ
sse . publish ( "status" , {
type: "connected" ,
userId ,
});
// μ£ΌκΈ°μ μΌλ‘ μ΄λ²€νΈ μ μ‘
const interval = setInterval (() => {
sse . publish ( "message" , {
id: Date . now (),
text: "New notification" ,
timestamp: new Date (),
});
}, 5000 );
// 5λΆ ν μ°κ²° μ’
λ£
setTimeout ( async () => {
clearInterval ( interval );
sse . publish ( "status" , {
type: "disconnected" ,
userId ,
});
await sse . end ();
}, 300000 );
return sse ;
}
}
μ€νΈλ¦¬λ° νμ
μ μ§μ ν©λλ€.
νμ μ΅μ
type : "sse" ; // Server-Sent Events (νμ¬ μ μΌν μ΅μ
)
WebSocketμ λ³λμ
@websocket λ°μ½λ μ΄ν°λ‘ μ§μλ©λλ€.
@streamμ SSE μ μ©μ΄λ©°
type μ΅μ
μ
"sse"λ§ λ°μ΅λλ€.
μ μ‘ν μ΄λ²€νΈμ νμ
μ Zod μ€ν€λ§λ‘ μ μν©λλ€.
νμ μ΅μ
import { z } from "zod" ;
const EventsSchema = z . object ({
// μ΄λ²€νΈ μ΄λ¦: μ΄λ²€νΈ λ°μ΄ν° μ€ν€λ§
progress: z . object ({
percent: z . number (),
message: z . string ()
}),
complete: z . object ({
result: z . string ()
}),
error: z . object ({
message: z . string (),
code: z . string ()
})
});
@ stream ({
type: "sse" ,
events: EventsSchema
})
async process () {
const sse = Sonamu . getContext (). createSSE ( EventsSchema );
// μ΄λ²€νΈ μ μ‘
sse . publish ( "progress" , { percent: 50 , message: "Processing..." });
sse . publish ( "complete" , { result: "Success" });
// μ€νΈλ¦Ό μ’
λ£
await sse . end ();
return sse ;
}
μ€νΈλ¦¬λ° μλν¬μΈνΈ κ²½λ‘λ₯Ό μ§μ ν©λλ€.
κΈ°λ³Έκ°: /{modelName}/{methodName} (camelCase)
@ stream ({
type: "sse" ,
events: EventsSchema ,
path: "/stream/notifications"
})
async subscribe () {
// /stream/notificationsλ‘ μ κ·Ό
}
μ κ·Ό κΆνμ μ§μ ν©λλ€.
type GuardKey = "query" | "admin" | "user" ;
@ stream ({
type: "sse" ,
events: EventsSchema ,
guards: [ "user" ]
})
async subscribe () {
// λ‘κ·ΈμΈν μ¬μ©μλ§ μ κ·Ό κ°λ₯
}
description
μ€νΈλ¦¬λ° API μ€λͺ
μ μΆκ°ν©λλ€.
@ stream ({
type: "sse" ,
events: EventsSchema ,
description: "μ€μκ° μλ¦Όμ μ€νΈλ¦¬λ°ν©λλ€."
})
async notifications () {
// ...
}
resourceName
μμ±λλ μλΉμ€ νμΌμ 리μμ€ μ΄λ¦μ μ§μ ν©λλ€.
@ stream ({
type: "sse" ,
events: EventsSchema ,
resourceName: "Notifications"
})
async subscribe () {
// NotificationsService.ts νμΌμ ν¬ν¨
}
SSE κ°μ²΄ μ¬μ©λ²
createSSE
Contextμμ SSE κ°μ²΄λ₯Ό μμ±ν©λλ€.
const sse = Sonamu . getContext (). createSSE ( EventsSchema );
publish
μ΄λ²€νΈλ₯Ό ν΄λΌμ΄μΈνΈλ‘ μ μ‘ν©λλ€.
sse . publish ( "eventName" , data );
νμ
μμ μ±:
μ΄λ²€νΈ μ΄λ¦μ μ€ν€λ§μ μ μλ κ²λ§ μ¬μ© κ°λ₯
λ°μ΄ν°λ μ€ν€λ§μ λ§μΆ° μλ κ²μ¦
const EventsSchema = z . object ({
message: z . object ({
text: z . string (),
}),
});
const sse = context . createSSE ( EventsSchema );
// β
μ¬λ°λ₯Έ μ¬μ©
sse . publish ( "message" , { text: "Hello" });
// β νμ
μλ¬
sse . publish ( "unknown" , { text: "Hello" }); // 'unknown'μ μ€ν€λ§μ μμ
sse . publish ( "message" , { text: 123 }); // textλ stringμ΄μ΄μΌ ν¨
μ€νΈλ¦Ό μ°κ²°μ μ’
λ£ν©λλ€.
end()λ ν΄λΌμ΄μΈνΈμ βendβ μ΄λ²€νΈλ₯Ό μ μ‘νκ³ 200ms λκΈ° ν μ°κ²°μ λ«μ΅λλ€.
μ 체 μμ
import { BaseModelClass , stream } from "sonamu" ;
import { z } from "zod" ;
const LogEventsSchema = z . object ({
log: z . object ({
level: z . enum ([ "info" , "warn" , "error" ]),
message: z . string (),
timestamp: z . string (),
}),
progress: z . object ({
current: z . number (),
total: z . number (),
}),
complete: z . object ({
success: z . boolean (),
duration: z . number (),
}),
});
class TaskModelClass extends BaseModelClass {
@ stream ({
type: "sse" ,
events: LogEventsSchema ,
path: "/tasks/logs" ,
guards: [ "admin" ],
description: "μμ
μ€ν λ‘κ·Έλ₯Ό μ€μκ°μΌλ‘ μ€νΈλ¦¬λ°ν©λλ€." ,
})
async streamLogs ( taskId : number ) {
const sse = Sonamu . getContext (). createSSE ( LogEventsSchema );
const startTime = Date . now ();
// μμ
μμ
sse . publish ( "log" , {
level: "info" ,
message: "Task started" ,
timestamp: new Date (). toISOString (),
});
// μμ
μ€ν
const total = 100 ;
for ( let i = 0 ; i <= total ; i ++ ) {
await this . processItem ( i );
// μ§νλ₯ μ μ‘
sse . publish ( "progress" , {
current: i ,
total ,
});
// λ‘κ·Έ μ μ‘
if ( i % 10 === 0 ) {
sse . publish ( "log" , {
level: "info" ,
message: `Processed ${ i } / ${ total } items` ,
timestamp: new Date (). toISOString (),
});
}
}
// μμ
μλ£
sse . publish ( "complete" , {
success: true ,
duration: Date . now () - startTime ,
});
// μ°κ²° μ’
λ£
await sse . end ();
return sse ;
}
private async processItem ( index : number ) {
// μμ
μ²λ¦¬
await new Promise (( resolve ) => setTimeout ( resolve , 50 ));
}
}
λ€λ₯Έ λ°μ½λ μ΄ν°μ ν¨κ» μ¬μ©
@transactional
@ stream ({ type: "sse" , events: EventsSchema })
@ transactional ()
async process () {
const sse = Sonamu . getContext (). createSSE ( EventsSchema );
const wdb = this . getDB ( "w" );
// νΈλμμ
λ΄μμ μ€ν
// SSE μ μ‘κ³Ό DB μμ
ν¨κ» μν
await sse . end ();
return sse ;
}
@streamμ λ¨Όμ , @transactionalμ λμ€μ μμ±νμΈμ.
μ μ½μ¬ν
1. λ€λ₯Έ λΌμ°ν
λ°μ½λ μ΄ν°(@api, @websocket, @upload)μ μ€λ³΅ μ¬μ© λΆκ°
κ°μ λ©μλμ @api, @stream, @websocket, @upload μ€ λ μ΄μμ ν¨κ» μ¬μ©ν μ μμ΅λλ€.
// β μλ¬ λ°μ
@ api ()
@ stream ({ type: "sse" , events: EventsSchema })
async subscribe () {}
μλ¬ λ©μμ§:
@stream decorator can only be used once on NotificationModel.subscribe.
You can use only one of @api, @stream, @websocket, or @upload decorator on the same method.
2. eventsλ νμ
// β μλ¬ λ°μ
@ stream ({ type: "sse" })
async subscribe () {}
// events μ΅μ
μ΄ νμν©λλ€
3. httpMethodλ νμ GET
@streamμ μλμΌλ‘ httpMethod: "GET"μ μ€μ ν©λλ€.
@ stream ({ type: "sse" , events: EventsSchema })
async subscribe () {
// GET /notification/subscribeλ‘ λ
ΈμΆλ¨
}
ν΄λΌμ΄μΈνΈ μ¬μ© (Web)
Sonamuλ μλμΌλ‘ SSE ν΄λΌμ΄μΈνΈ μ½λλ₯Ό μμ±ν©λλ€.
// μλ μμ±λ μ½λ
import { NotificationService } from "@/services/NotificationService" ;
// SSE μ°κ²°
const eventSource = NotificationService . subscribe ( userId );
// μ΄λ²€νΈ 리μ€λ λ±λ‘
eventSource . addEventListener ( "message" , ( event ) => {
const data = JSON . parse ( event . data );
console . log ( "New message:" , data );
});
eventSource . addEventListener ( "status" , ( event ) => {
const data = JSON . parse ( event . data );
console . log ( "Status:" , data );
});
// μ°κ²° μ’
λ£ μ΄λ²€νΈ
eventSource . addEventListener ( "end" , () => {
console . log ( "Stream ended" );
eventSource . close ();
});
// μλ μ°κ²° μ’
λ£
eventSource . close ();
React μμ
import { useEffect , useState } from "react" ;
function NotificationList ({ userId } : { userId : number }) {
const [ messages , setMessages ] = useState < Message []>([]);
const [ status , setStatus ] = useState < string >( "disconnected" );
useEffect (() => {
const eventSource = NotificationService . subscribe ( userId );
eventSource . addEventListener ( "message" , ( event ) => {
const message = JSON . parse ( event . data );
setMessages ( prev => [ ... prev , message ]);
});
eventSource . addEventListener ( "status" , ( event ) => {
const { type } = JSON . parse ( event . data );
setStatus ( type );
});
eventSource . addEventListener ( "end" , () => {
console . log ( "Stream ended by server" );
setStatus ( "disconnected" );
});
return () => {
eventSource . close ();
};
}, [ userId ]);
return (
< div >
< div > Status : { status }</ div >
< ul >
{ messages . map ( msg => (
< li key = {msg. id } > {msg. text } </ li >
))}
</ ul >
</ div >
);
}
@stream λ°μ½λ μ΄ν°λ μλμΌλ‘ λ‘κ·Έλ₯Ό λ¨κΉλλ€:
@ stream ({ type: "sse" , events: EventsSchema })
async subscribe () {
// μλ λ‘κ·Έ:
// [DEBUG] stream: NotificationModel.subscribe
}
μμ λͺ¨μ
μ±ν
μ§νλ₯ λͺ¨λν°λ§
μ€μκ° λ°μ΄ν°
const ChatEventsSchema = z . object ({
message: z . object ({
id: z . string (),
userId: z . number (),
text: z . string (),
timestamp: z . date ()
}),
typing: z . object ({
userId: z . number (),
isTyping: z . boolean ()
}),
joined: z . object ({
userId: z . number (),
username: z . string ()
}),
left: z . object ({
userId: z . number ()
})
});
class ChatModelClass extends BaseModelClass {
@ stream ({
type: "sse" ,
events: ChatEventsSchema ,
guards: [ "user" ]
})
async subscribe ( roomId : number ) {
const sse = Sonamu . getContext (). createSSE ( ChatEventsSchema );
const { user } = Sonamu . getContext ();
// μ
μ₯ μλ¦Ό
sse . publish ( "joined" , {
userId: user . id ,
username: user . name
});
// λ©μμ§ μμ 리μ€λ λ±λ‘
const unsubscribe = this . onMessage ( roomId , ( message ) => {
sse . publish ( "message" , message );
});
// 1μκ° ν μλ μ’
λ£
setTimeout ( async () => {
unsubscribe ();
sse . publish ( "left" , { userId: user . id });
await sse . end ();
}, 3600000 );
return sse ;
}
}
const ProcessEventsSchema = z . object ({
started: z . object ({
taskId: z . string (),
totalSteps: z . number ()
}),
progress: z . object ({
step: z . number (),
message: z . string (),
percent: z . number ()
}),
completed: z . object ({
success: z . boolean (),
result: z . any (),
duration: z . number ()
}),
error: z . object ({
step: z . number (),
message: z . string ()
})
});
class ProcessModelClass extends BaseModelClass {
@ stream ({
type: "sse" ,
events: ProcessEventsSchema ,
guards: [ "admin" ]
})
async monitor ( processId : string ) {
const sse = Sonamu . getContext (). createSSE ( ProcessEventsSchema );
const startTime = Date . now ();
try {
const steps = await this . getProcessSteps ( processId );
sse . publish ( "started" , {
taskId: processId ,
totalSteps: steps . length
});
for ( let i = 0 ; i < steps . length ; i ++ ) {
const step = steps [ i ];
sse . publish ( "progress" , {
step: i + 1 ,
message: step . description ,
percent: (( i + 1 ) / steps . length ) * 100
});
await this . executeStep ( step );
}
sse . publish ( "completed" , {
success: true ,
result: await this . getProcessResult ( processId ),
duration: Date . now () - startTime
});
} catch ( error ) {
sse . publish ( "error" , {
step: - 1 ,
message: error . message
});
}
await sse . end ();
return sse ;
}
}
const StockEventsSchema = z . object ({
price: z . object ({
symbol: z . string (),
price: z . number (),
change: z . number (),
timestamp: z . date ()
}),
volume: z . object ({
symbol: z . string (),
volume: z . number ()
}),
alert: z . object ({
type: z . enum ([ "high" , "low" ]),
symbol: z . string (),
message: z . string ()
})
});
class StockModelClass extends BaseModelClass {
@ stream ({
type: "sse" ,
events: StockEventsSchema ,
guards: [ "user" ]
})
async watchSymbol ( symbol : string ) {
const sse = Sonamu . getContext (). createSSE ( StockEventsSchema );
// 1μ΄λ§λ€ κ°κ²© μ
λ°μ΄νΈ (1λΆκ°)
for ( let i = 0 ; i < 60 ; i ++ ) {
const quote = await this . getQuote ( symbol );
sse . publish ( "price" , {
symbol ,
price: quote . price ,
change: quote . change ,
timestamp: new Date ()
});
// κ°κ²© μλ¦Ό
if ( quote . price > quote . highThreshold ) {
sse . publish ( "alert" , {
type: "high" ,
symbol ,
message: `Price exceeded ${ quote . highThreshold } `
});
}
await new Promise ( resolve => setTimeout ( resolve , 1000 ));
}
await sse . end ();
return sse ;
}
}
λ€μ λ¨κ³
@api μΌλ° API μλν¬μΈνΈ λ§λ€κΈ°
@transactional νΈλμμ
μΌλ‘ λ°μ΄ν° μΌκ΄μ± 보μ₯νκΈ°
@cache κ²°κ³Ό μΊμ±μΌλ‘ μ±λ₯ ν₯μνκΈ°
SSE κ°μ΄λ SSE μμΈ κ°μ΄λ 보기